Make MergeIntoTestUtils agnostic to path/name-based access

Author: ctring

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

@@ -624,6 +624,14 @@ trait DeltaDMLTestUtils
     }
   }
 
+  /**
+   * Reads a delta table by its identifier. The identifier can either be the table name or table
+   * path that is in the form of delta.`tablePath`.
+   */
+  protected def readDeltaTableByIdentifier(tableIdentifier: String): DataFrame = {
+    spark.read.format("delta").table(tableIdentifier)
+  }
+
   /**
    * 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.

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

@@ -72,19 +72,19 @@ trait MergeIntoSchemaEvolutionMixin {
 
         if (error != null) {
           val ex = intercept[AnalysisException] {
-            executeMerge(s"delta.`$tempPath` t", "source s", cond, clauses: _*)
+            executeMerge(s"$tableSQLIdentifier t", "source s", cond, clauses: _*)
           }
           errorContains(Utils.exceptionString(ex), error)
         } else {
-          executeMerge(s"delta.`$tempPath` t", "source s", cond, clauses: _*)
-          checkAnswer(spark.read.format("delta").load(tempPath), df.collect())
+          executeMerge(s"$tableSQLIdentifier t", "source s", cond, clauses: _*)
+          checkAnswer(readDeltaTableByIdentifier(tableSQLIdentifier), df.collect())
           if (schema != null) {
-            assert(spark.read.format("delta").load(tempPath).schema === schema)
+            assert(readDeltaTableByIdentifier(tableSQLIdentifier).schema === schema)
           } else {
             // Check against the schema of the expected result df if no explicit schema was
             // provided. Nullability of fields will vary depending on the actual data in the df so
             // we ignore it.
-            assert(spark.read.format("delta").load(tempPath).schema.asNullable ===
+            assert(readDeltaTableByIdentifier(tableSQLIdentifier).schema.asNullable ===
               df.schema.asNullable)
           }
         }

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

@@ -45,6 +45,7 @@ abstract class MergeIntoSuiteBase
     extends QueryTest
     with SharedSparkSession
     with DeltaSQLTestUtils
+    with DeltaDMLByPathTestUtils
     with ScanReportHelper
     with DeltaTestUtilsForTempViews
     with MergeIntoTestUtils

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 DeltaDMLByPathTestUtils with MergeHelpers {
+trait MergeIntoTestUtils extends DeltaDMLTestUtils with MergeHelpers {
   self: SharedSparkSession =>
 
   protected def executeMerge(

spark/src/test/scala/org/apache/spark/sql/delta/cdc/MergeCDCSuite.scala

@@ -44,6 +44,7 @@ class MergeCDCSuite extends MergeIntoSQLSuite with MergeCDCTests
  */
 trait MergeCDCTests extends QueryTest
   with MergeIntoSQLTestUtils
+  with DeltaDMLByPathTestUtils
   with DeltaColumnMappingTestUtils
   with DeltaSQLCommandTest {
   import testImplicits._

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

@@ -38,7 +38,6 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
  */
 class TypeWideningInsertSchemaEvolutionBasicSuite
     extends QueryTest
-    with DeltaDMLTestUtils
     with TypeWideningTestMixin
     with TypeWideningInsertSchemaEvolutionBasicTests {