Revert to unsorted Dataframe for Delta and Hudi

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/processing/LakeWriter.scala

@@ -155,7 +155,7 @@ object LakeWriter {
 
     def commit(viewName: String): F[Unit] =
       for {
-        df <- SparkUtils.prepareFinalDataFrame(spark, viewName, writerParallelism)
+        df <- SparkUtils.prepareFinalDataFrame(spark, viewName, writerParallelism, w.expectsSortedDataframe)
         _ <- mutex.lock
                .surround {
                  w.write(df)

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/processing/SparkUtils.scala

@@ -103,15 +103,18 @@ private[processing] object SparkUtils {
   def prepareFinalDataFrame[F[_]: Sync](
     spark: SparkSession,
     viewName: String,
-    writerParallelism: Int
+    writerParallelism: Int,
+    writerExpectsSortedDataframe: Boolean
   ): F[DataFrame] =
-    Sync[F].delay {
-      spark
-        .table(viewName)
-        .repartitionByRange(writerParallelism, col("event_name"), col("event_id"))
-        .sortWithinPartitions("event_name")
-        .withColumn("load_tstamp", current_timestamp())
-    }
+    for {
+      df <- Sync[F].pure(spark.table(viewName))
+      df <- Sync[F].pure {
+              // Create equally-balanced partitions, for which events with similar event_name are likely to be in the same partition.
+              // This maximizes output file sizes, for a lake which is partitioned by event_name.
+              if (writerParallelism > 1) df.repartitionByRange(writerParallelism, col("event_name"), col("event_id")) else df.coalesce(1)
+            }
+      df <- Sync[F].pure(if (writerExpectsSortedDataframe) df.sortWithinPartitions("event_name") else df)
+    } yield df.withColumn("load_tstamp", current_timestamp())
 
   def dropView[F[_]: Sync](spark: SparkSession, viewName: String): F[Unit] =
     Logger[F].info(s"Removing Spark data frame $viewName from local disk...") >>

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/DeltaWriter.scala

@@ -111,4 +111,5 @@ class DeltaWriter(config: Config.Delta) extends Writer {
    */
   override def toleratesAsyncDelete: Boolean = true
 
+  override def expectsSortedDataframe: Boolean = false
 }

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/HudiWriter.scala

@@ -94,4 +94,6 @@ class HudiWriter(config: Config.Hudi) extends Writer {
    * steps must happen in order.
    */
   override def toleratesAsyncDelete: Boolean = false
+
+  override def expectsSortedDataframe: Boolean = false
 }

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/IcebergWriter.scala

@@ -112,4 +112,10 @@ class IcebergWriter(config: Config.Iceberg) extends Writer {
    * re-writes a file that was previously deleted
    */
   override def toleratesAsyncDelete: Boolean = true
+
+  /**
+   * Iceberg writer requires the Dataframe to be sorted, because we set the iceberg write option
+   * `distribution-mode = none`
+   */
+  override def expectsSortedDataframe: Boolean = true
 }

modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/Writer.scala

@@ -36,4 +36,10 @@ trait Writer {
    * If tolerated, then we use our customized `LakeLoaderFileSystem`.
    */
   def toleratesAsyncDelete: Boolean
+
+  /**
+   * Whether this writer expects the DataFrame to be sorted by the partition column, i.e. by
+   * event_name
+   */
+  def expectsSortedDataframe: Boolean
 }