[!458] - Resolve "Parallel Writer plugin tests are very slow"

# New features and improvements

- Running `sbt wasp-plugin-parallel-write-spark/test` takes 90s on a M1 Max MBP.

# Breaking changes

None.

# Migration

None.

# Bug fixes

None.

# How this feature was tested

Existing tests

# Related issue

Closes #574

Author: agilelab-tmnd1991

plugin-parallel-write-spark/src/test/scala/it/agilelab/bigdata/wasp/consumers/spark/plugins/parallel/tools/ContinuousUpdateWriterSpec.scala

@@ -6,15 +6,21 @@ import it.agilelab.bigdata.wasp.consumers.spark.plugins.parallel.tools.utils.Par
 import org.apache.spark.sql.execution.streaming.MemoryStream
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.streaming.StreamingQueryException
-import org.apache.spark.sql.types.{ DataType, StructField, StructType }
+import org.apache.spark.sql.types.{DataType, StructField, StructType}
 import org.scalatest.FunSuite
-import org.scalatest.Matchers.{ an, be }
+import org.scalatest.Matchers.{an, be}
 
 case class Schema(ordering: Int, column1: String, column2: String)
 case class Schema2(ordering: Int, column1: String, column2: String, column3: String)
 case class NotSupportedSchema(ordering: Int, column1: String)
 case class Schema3(ordering1: Int, ordering2: Float, column1: String, column2: String)
-case class Schema4(ordering1: Int, ordering2: Float, column1: String, column2: String, colThatIsNotPartOfDeltaTableSchema: String)
+case class Schema4(
+    ordering1: Int,
+    ordering2: Float,
+    column1: String,
+    column2: String,
+    colThatIsNotPartOfDeltaTableSchema: String
+)
 case class CaseSensitiveTest(orDeRing: Int, column1: String, column2: String)
 
 class ContinuousUpdateWriterSpec extends FunSuite with DeltaTableTest {
@@ -316,7 +322,9 @@ class ContinuousUpdateWriterSpec extends FunSuite with DeltaTableTest {
     }
   }
 
-  test("Data deduplication when enforcing schema involves dropping columns and the result of the ordering expression is ambiguous") {
+  test(
+    "Data deduplication when enforcing schema involves dropping columns and the result of the ordering expression is ambiguous"
+  ) {
     withServer(dispatcher) { serverData =>
       val data: Seq[Schema4] = Seq(
         Schema4(1, 0.4f, "key1", "value1", "thisColWillBeDroppedBeforeWriting1"),

plugin-parallel-write-spark/src/test/scala/it/agilelab/bigdata/wasp/consumers/spark/plugins/parallel/tools/TestModels.scala

@@ -1,24 +1,69 @@
 package it.agilelab.bigdata.wasp.consumers.spark.plugins.parallel.tools
 
 import it.agilelab.bigdata.wasp.consumers.spark.plugins.parallel.catalog.CatalogCoordinates
-import it.agilelab.bigdata.wasp.consumers.spark.plugins.parallel.model.{ContinuousUpdate, ParallelWrite, ParallelWriteModel}
+import it.agilelab.bigdata.wasp.consumers.spark.plugins.parallel.model.{
+  ContinuousUpdate,
+  ParallelWrite,
+  ParallelWriteModel
+}
 
 object TestModels {
   val model1 = ParallelWriteModel(ParallelWrite("append"), entityDetails = CatalogCoordinates("default", "mock", "v1"))
-  val model2 = ParallelWriteModel(ParallelWrite("overwrite"), entityDetails = CatalogCoordinates("default", "mock", "v1"))
+  val model2 =
+    ParallelWriteModel(ParallelWrite("overwrite"), entityDetails = CatalogCoordinates("default", "mock", "v1"))
 
-  private val entityDetails: CatalogCoordinates = CatalogCoordinates("default", "mock", "v1")
-  private val entityDetailsWithDB: CatalogCoordinates = CatalogCoordinates("default", "mock", "v1", Some("mock_db"))
-  private val wrongDetails: CatalogCoordinates = CatalogCoordinates("default", "fake", "v1")
+  private val entityDetails: CatalogCoordinates            = CatalogCoordinates("default", "mock", "v1")
+  private val entityDetailsWithDB: CatalogCoordinates      = CatalogCoordinates("default", "mock", "v1", Some("mock_db"))
+  private val wrongDetails: CatalogCoordinates             = CatalogCoordinates("default", "fake", "v1")
   private val notExistingEntityDetails: CatalogCoordinates = CatalogCoordinates("default", "entity", "v1")
 
   val modelWithDB = ParallelWriteModel(ParallelWrite("overwrite"), entityDetails = entityDetailsWithDB)
 
-  val continuousUpdateModel1 = ParallelWriteModel(ContinuousUpdate(keys = "column1" :: Nil, orderingExpression = "ordering", Some(100), Some(1), Some(168), Some(100)), entityDetails)
+  val continuousUpdateModel1 = ParallelWriteModel(
+    ContinuousUpdate(
+      keys = "column1" :: Nil,
+      orderingExpression = "ordering",
+      compactFrequency = None,
+      compactNumFile = None,
+      retentionHours = None,
+      vacuumFrequency = None
+    ),
+    entityDetails
+  )
 
-  val continuousUpdateModel2 = ParallelWriteModel(ContinuousUpdate(keys = "column1" :: Nil, orderingExpression = "-(ordering1 + ordering2)", Some(100), Some(1), Some(168), Some(100)), entityDetails)
+  val continuousUpdateModel2 = ParallelWriteModel(
+    ContinuousUpdate(
+      keys = "column1" :: Nil,
+      orderingExpression = "-(ordering1 + ordering2)",
+      compactFrequency = Some(100),
+      compactNumFile = Some(1),
+      retentionHours = Some(168),
+      vacuumFrequency = Some(100)
+    ),
+    entityDetails
+  )
 
-  val wrongModel = ParallelWriteModel(ContinuousUpdate(keys = "column1" :: Nil, orderingExpression = "-(ordering1 + ordering2)", Some(100), Some(1), Some(168), Some(100)), wrongDetails)
+  val wrongModel = ParallelWriteModel(
+    ContinuousUpdate(
+      keys = "column1" :: Nil,
+      orderingExpression = "-(ordering1 + ordering2)",
+      compactFrequency = Some(100),
+      compactNumFile = Some(1),
+      retentionHours = Some(168),
+      vacuumFrequency = Some(100)
+    ),
+    wrongDetails
+  )
 
-  val notExistingEntityModel = ParallelWriteModel(ContinuousUpdate(keys = "column1" :: Nil, orderingExpression = "-(ordering1 + ordering2)", Some(100), Some(1), Some(168), Some(100)), notExistingEntityDetails)
+  val notExistingEntityModel = ParallelWriteModel(
+    ContinuousUpdate(
+      keys = "column1" :: Nil,
+      orderingExpression = "-(ordering1 + ordering2)",
+      compactFrequency = Some(100),
+      compactNumFile = Some(1),
+      retentionHours = Some(168),
+      vacuumFrequency = Some(100)
+    ),
+    notExistingEntityDetails
+  )
 }