Delta Test Suite Refactoring

Closes delta-io#2292 GitOrigin-RevId: 27e88f1e2dc5c8b8dad5ea36fa348476dd6f3d6c
andreaschat-db · Nov 20, 2023 · 35e5d69 · 35e5d69
1 parent 246fcf3
commit 35e5d69
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 16 deletions.
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceDeletionVectorsSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceDeletionVectorsSuite.scala
@@ -195,7 +195,6 @@ trait DeltaSourceDeletionVectorTests extends StreamTest
        } else {
          Seq(
            // This makes it move to the next batch.
-           AssertOnQuery(waitUntilBatchProcessed(1, _)),
            AssertOnQuery { q =>
              eventually("Next batch was never processed") {
                // Ensure we only processed a single batch with the DML commands.
@@ -208,18 +207,6 @@ trait DeltaSourceDeletionVectorTests extends StreamTest
     testStream(df)((baseActions ++ expectations): _*)
   }
 
-  protected def waitUntilBatchProcessed(batchId: Int, currentStream: StreamExecution): Boolean = {
-    eventually("Next batch was never processed") {
-      if (!currentStream.exception.isDefined) {
-        assert(currentStream.commitLog.getLatestBatchId().get >= batchId)
-      }
-    }
-    if (currentStream.exception.isDefined) {
-      throw currentStream.exception.get
-    }
-    true
-  }
-
   protected def eventually[T](message: String)(func: => T): T = {
     try {
       Eventually.eventually(Timeout(streamingTimeout)) {

diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
@@ -142,8 +142,9 @@ trait StreamingSchemaEvolutionSuiteBase extends ColumnMappingStreamingTestUtils
 
   protected def testSchemaEvolution(
       testName: String,
-      columnMapping: Boolean = true)(f: DeltaLog => Unit): Unit = {
-    super.test(testName) {
+      columnMapping: Boolean = true,
+      tags: Seq[org.scalatest.Tag] = Seq.empty)(f: DeltaLog => Unit): Unit = {
+    super.test(testName, tags: _*) {
       if (columnMapping) {
         withStarterTable { log =>
           f(log)
@@ -1415,7 +1416,8 @@ trait StreamingSchemaEvolutionSuiteBase extends ColumnMappingStreamingTestUtils
     }
   }
 
-  testSchemaEvolution("multiple sources with schema evolution") { implicit log =>
+  testSchemaEvolution("multiple sources with schema evolution"
+    ) { implicit log =>
     val v5 = log.update().version // v5 has an ADD file action with value (4, 4)
     renameColumn("b", "c")
     addData(5 until 10)