Support SeparateLines output mode for assertSmallDatasetEquality (#213)

Author: zeotuan

core/src/main/scala/com/github/mrpowers/spark/fast/tests/DataFrameComparer.scala

@@ -22,6 +22,8 @@ trait DataFrameComparer extends DatasetComparer {
    *   \- don't compare column metadata when matching schemas
    * @param truncate
    *   \- truncate column if length more than specified number
+   * @param outputFormat
+   *   \- format of the dataframe diff output either SideBySide or SeparateLines
    */
   def assertSmallDataFrameEquality(
       actualDF: DataFrame,
@@ -34,51 +36,17 @@ trait DataFrameComparer extends DatasetComparer {
       truncate: Int = 500,
       outputFormat: DataframeDiffOutputFormat = DataframeDiffOutputFormat.SideBySide
   ): Unit = {
-    outputFormat match {
-      case DataframeDiffOutputFormat.SideBySide =>
-        assertSmallDatasetEquality(
-          actualDF,
-          expectedDF,
-          ignoreNullable,
-          ignoreColumnNames,
-          orderedComparison,
-          ignoreColumnOrder,
-          ignoreMetadata,
-          truncate
-        )
-      case DataframeDiffOutputFormat.SeparateLines =>
-        SchemaComparer.assertSchemaEqual(
-          actualDF.schema,
-          expectedDF.schema,
-          ignoreNullable,
-          ignoreColumnNames,
-          ignoreColumnOrder,
-          ignoreMetadata
-        )
-        val actual = if (ignoreColumnOrder) orderColumns(actualDF, expectedDF) else actualDF
-        if (orderedComparison)
-          assertSmallDataFrameEquality(actual, expectedDF, truncate)
-        else
-          assertSmallDataFrameEquality(
-            defaultSortDataset(actual),
-            defaultSortDataset(expectedDF),
-            truncate
-          )
-    }
-
-  }
-
-  private def assertSmallDataFrameEquality(
-      actualDF: DataFrame,
-      expectedDF: DataFrame,
-      truncate: Int
-  ): Unit = {
-    val a = actualDF.collect()
-    val e = expectedDF.collect()
-    if (!a.toSeq.approximateSameElements(e, (o1: Row, o2: Row) => o1.equals(o2))) {
-      val msg = "Difference\n" ++ DataframeUtil.showDataframeDiff(a, e, actualDF.schema.fieldNames, truncate)
-      throw DatasetContentMismatch(msg)
-    }
+    assertSmallDatasetEquality(
+      actualDF,
+      expectedDF,
+      ignoreNullable,
+      ignoreColumnNames,
+      orderedComparison,
+      ignoreColumnOrder,
+      ignoreMetadata,
+      truncate,
+      outputFormat = outputFormat
+    )
   }
 
   /**

core/src/main/scala/com/github/mrpowers/spark/fast/tests/DataframeUtil.scala

@@ -1,5 +1,6 @@
 package com.github.mrpowers.spark.fast.tests
 
+import com.github.mrpowers.spark.fast.tests.DataframeDiffOutputFormat.DataframeDiffOutputFormat
 import com.github.mrpowers.spark.fast.tests.DatasetComparer.maxUnequalRowsToShow
 import com.github.mrpowers.spark.fast.tests.SeqLikesExtensions.SeqExtensions
 import org.apache.spark.rdd.RDD
@@ -47,32 +48,39 @@ Expected DataFrame Row Count: '$expectedCount'
       ignoreColumnOrder: Boolean = false,
       ignoreMetadata: Boolean = true,
       truncate: Int = 500,
-      equals: (T, T) => Boolean = (o1: T, o2: T) => o1.equals(o2)
+      equals: (T, T) => Boolean = (o1: T, o2: T) => o1.equals(o2),
+      outputFormat: DataframeDiffOutputFormat = DataframeDiffOutputFormat.SideBySide
   ): Unit = {
     SchemaComparer.assertDatasetSchemaEqual(actualDS, expectedDS, ignoreNullable, ignoreColumnNames, ignoreColumnOrder, ignoreMetadata)
     val actual = if (ignoreColumnOrder) orderColumns(actualDS, expectedDS) else actualDS
-    assertSmallDatasetContentEquality(actual, expectedDS, orderedComparison, truncate, equals)
+    assertSmallDatasetContentEquality(actual, expectedDS, orderedComparison, truncate, equals, outputFormat)
   }
 
   def assertSmallDatasetContentEquality[T: ClassTag](
       actualDS: Dataset[T],
       expectedDS: Dataset[T],
       orderedComparison: Boolean,
       truncate: Int,
-      equals: (T, T) => Boolean
+      equals: (T, T) => Boolean,
+      outputFormat: DataframeDiffOutputFormat
   ): Unit = {
     if (orderedComparison)
-      assertSmallDatasetContentEquality(actualDS, expectedDS, truncate, equals)
+      assertSmallDatasetContentEquality(actualDS, expectedDS, truncate, equals, outputFormat)
     else
-      assertSmallDatasetContentEquality(defaultSortDataset(actualDS), defaultSortDataset(expectedDS), truncate, equals)
+      assertSmallDatasetContentEquality(defaultSortDataset(actualDS), defaultSortDataset(expectedDS), truncate, equals, outputFormat)
   }
 
-  def assertSmallDatasetContentEquality[T: ClassTag](actualDS: Dataset[T], expectedDS: Dataset[T], truncate: Int, equals: (T, T) => Boolean): Unit = {
+  def assertSmallDatasetContentEquality[T: ClassTag](
+      actualDS: Dataset[T],
+      expectedDS: Dataset[T],
+      truncate: Int,
+      equals: (T, T) => Boolean,
+      outputFormat: DataframeDiffOutputFormat
+  ): Unit = {
     val a = actualDS.collect().toSeq
     val e = expectedDS.collect().toSeq
     if (!a.approximateSameElements(e, equals)) {
-      val arr = ("Actual Content", "Expected Content")
-      val msg = "Diffs\n" ++ ProductUtil.showProductDiff(arr, a, e, truncate)
+      val msg = "Diffs\n" ++ ProductUtil.showProductDiff(expectedDS.columns, a, e, truncate, outputFormat = outputFormat)
       throw DatasetContentMismatch(msg)
     }
   }