[Spark] Make DeltaDMLTestUtils agnostic to name/path-based access (delta-io#4738)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

This PR is one of a series of PRs aiming to refactor the DML suites such
that we can extend the existing suites to test with name-based table
access.

Currently, all DML suites are centered around the `DeltaDMLTestUtils`
trait that creates a test table stored in `tempPath`. In this PR, all
path-related elements in this trait will be abstracted out and put into
a new specialized trait called `DeltaDMLByPathTestUtils`.

## How was this patch tested?

Existing UTs 

## Does this PR introduce _any_ user-facing changes?

No

Author: ctring

spark/src/test/scala/org/apache/spark/sql/delta/DeleteSuiteBase.scala

@@ -27,7 +27,7 @@ import org.apache.spark.sql.types.StructType
 
 abstract class DeleteSuiteBase extends QueryTest
   with SharedSparkSession
-  with DeltaDMLTestUtils
+  with DeltaDMLByPathTestUtils
   with DeltaTestUtilsForTempViews
   with DeltaExcludedBySparkVersionTestMixinShims {
 

spark/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTest.scala

@@ -37,7 +37,10 @@ import org.apache.spark.sql.types.StructType
  * Each take a unique path through analysis. The abstractions below captures these different
  * inserts to allow more easily running tests with all or a subset of them.
  */
-trait DeltaInsertIntoTest extends QueryTest with DeltaDMLTestUtils with DeltaSQLCommandTest {
+trait DeltaInsertIntoTest
+  extends QueryTest
+  with DeltaDMLByPathTestUtils
+  with DeltaSQLCommandTest {
 
   val catalogName = "spark_catalog"
 

spark/src/test/scala/org/apache/spark/sql/delta/DeltaTestUtils.scala

@@ -568,34 +568,25 @@ trait DeltaDMLTestUtils
 
   import testImplicits._
 
-  protected var tempDir: File = _
-
   protected var deltaLog: DeltaLog = _
 
-  protected def tempPath: String = tempDir.getCanonicalPath
-
-  override protected def beforeEach(): Unit = {
-    super.beforeEach()
-    // Using a space in path to provide coverage for special characters.
-    tempDir = Utils.createTempDir(namePrefix = "spark test")
-    deltaLog = DeltaLog.forTable(spark, new Path(tempPath))
-  }
-
-  override protected def afterEach(): Unit = {
-    try {
-      Utils.deleteRecursively(tempDir)
-      DeltaLog.clearCache()
-    } finally {
-      super.afterEach()
-    }
-  }
+  protected def tableSQLIdentifier: String
 
   protected def append(df: DataFrame, partitionBy: Seq[String] = Nil): Unit = {
+    import DeltaTestUtils.TableIdentifierOrPath
+
     val dfw = df.write.format("delta").mode("append")
     if (partitionBy.nonEmpty) {
       dfw.partitionBy(partitionBy: _*)
     }
-    dfw.save(tempPath)
+    getTableIdentifierOrPath(tableSQLIdentifier) match {
+      case TableIdentifierOrPath.Identifier(id, _) => dfw.saveAsTable(id.toString)
+      // A cleaner way to write this is to just use `saveAsTable` where the
+      // table name is delta.`path`. However, it will throw an error when
+      // we use "append" mode and the table does not exist, so we use `save`
+      // here instead.
+      case TableIdentifierOrPath.Path(path, _) => dfw.save(path)
+    }
   }
 
   protected def withKeyValueData(
@@ -612,7 +603,7 @@ trait DeltaDMLTestUtils
       if (isKeyPartitioned) Seq(targetKeyValueNames._1) else Nil)
     withTempView("source") {
       source.toDF(sourceKeyValueNames._1, sourceKeyValueNames._2).createOrReplaceTempView("source")
-      thunk("source", s"delta.`$tempPath`")
+      thunk("source", tableSQLIdentifier)
     }
   }
 
@@ -633,12 +624,6 @@ trait DeltaDMLTestUtils
     }
   }
 
-  protected def readDeltaTable(path: String): DataFrame = {
-    spark.read.format("delta").load(path)
-  }
-
-  protected def getDeltaFileStmt(path: String): String = s"SELECT * FROM delta.`$path`"
-
   /**
    * Finds the latest operation of the given type that ran on the test table and returns the
    * dataframe with the changes of the corresponding table version.
@@ -665,3 +650,35 @@ trait DeltaDMLTestUtils
       .drop(CDCReader.CDC_COMMIT_VERSION)
   }
 }
+
+trait DeltaDMLByPathTestUtils extends DeltaDMLTestUtils {
+  self: SharedSparkSession =>
+
+  protected var tempDir: File = _
+
+  protected def tempPath: String = tempDir.getCanonicalPath
+
+  override protected def beforeEach(): Unit = {
+    super.beforeEach()
+    // Using a space in path to provide coverage for special characters.
+    tempDir = Utils.createTempDir(namePrefix = "spark test")
+    deltaLog = DeltaLog.forTable(spark, tempPath)
+  }
+
+  override protected def afterEach(): Unit = {
+    try {
+      Utils.deleteRecursively(tempDir)
+      DeltaLog.clearCache()
+    } finally {
+      super.afterEach()
+    }
+  }
+
+  override protected def tableSQLIdentifier: String = s"delta.`$tempPath`"
+
+  protected def readDeltaTable(path: String): DataFrame = {
+    spark.read.format("delta").load(path)
+  }
+
+  protected def getDeltaFileStmt(path: String): String = s"SELECT * FROM delta.`$path`"
+}

spark/src/test/scala/org/apache/spark/sql/delta/MergeIntoTestUtils.scala

@@ -28,7 +28,7 @@ import org.apache.spark.sql.test.SharedSparkSession
  * either [[MergeIntoSQLTestUtils]] or [[MergeIntoScalaTestUtils]] to run merge tests using the SQL
  * or Scala API resp.
  */
-trait MergeIntoTestUtils extends DeltaDMLTestUtils with MergeHelpers {
+trait MergeIntoTestUtils extends DeltaDMLByPathTestUtils with MergeHelpers {
   self: SharedSparkSession =>
 
   protected def executeMerge(

spark/src/test/scala/org/apache/spark/sql/delta/UpdateSuiteBase.scala

@@ -36,7 +36,7 @@ import org.apache.spark.sql.types._
 abstract class UpdateSuiteBase
   extends QueryTest
   with SharedSparkSession
-  with DeltaDMLTestUtils
+  with DeltaDMLByPathTestUtils
   with DeltaSQLTestUtils
   with DeltaTestUtilsForTempViews
   with DeltaExcludedBySparkVersionTestMixinShims {

spark/src/test/scala/org/apache/spark/sql/delta/typewidening/TypeWideningTestMixin.scala

@@ -35,7 +35,9 @@ import org.apache.spark.sql.types._
 /**
  * Test mixin that enables type widening by default for all tests in the suite.
  */
-trait TypeWideningTestMixin extends DeltaSQLCommandTest with DeltaDMLTestUtils { self: QueryTest =>
+trait TypeWideningTestMixin
+  extends DeltaSQLCommandTest
+  with DeltaDMLByPathTestUtils { self: QueryTest =>
 
   import testImplicits._