Use repartition(1) instead of coalesce(1) for OPTIMIZE

Signed-off-by: Eunjin Song <sezruby@gmail.com>
jbguerraz · Jul 6, 2022 · 4561420 · 4561420
1 parent 54a3871
commit 4561420
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 1 deletion.
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala
@@ -369,7 +369,13 @@ class OptimizeExecutor(
         approxNumFiles,
         zOrderByColumns)
     } else {
-      input.coalesce(numPartitions = 1)
+      val useRepartition = sparkSession.sessionState.conf.getConf(
+        DeltaSQLConf.DELTA_OPTIMIZE_USE_REPARTITON)
+      if (useRepartition) {
+        input.repartition(numPartitions = 1)
+      } else {
+        input.coalesce(numPartitions = 1)
+      }
     }
 
     val partitionDesc = partition.toSeq.map(entry => entry._1 + "=" + entry._2).mkString(",")

diff --git a/core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala
@@ -766,6 +766,12 @@ trait DeltaSQLConfBase {
       .transform(_.toLowerCase(Locale.ROOT))
       .createWithDefault("table")
 
+  val DELTA_OPTIMIZE_USE_REPARTITON =
+    buildConf("optimize.repartition.enabled")
+      .internal()
+      .doc("Use repartition(1) instead of coalesce(1) to merge small files.")
+      .booleanConf
+      .createWithDefault(false)
 
   val DELTA_ALTER_TABLE_CHANGE_COLUMN_CHECK_EXPRESSIONS =
     buildConf("alterTable.changeColumn.checkExpressions")