[Spark] add auto-compact post commit hook #2414

nicklan · 2023-12-28T19:32:38Z

Which Delta project/connector is this regarding?

Description

Adds code to support Auto Compaction.

Auto compaction combines small files within partitions to reduce problems due to a proliferation of small files. Auto compaction is implemented as a post commit hook, and so occurs after the write to a table has succeeded. It runs synchronously on the cluster that has performed the write.

You can control the output file size by setting the spark.databricks.delta.autoCompact.maxFileSize.

Auto compaction is only triggered for partitions or tables that have at least a certain number of small files. You can optionally change the minimum number of files required to trigger auto compaction by setting spark.databricks.delta.autoCompact.minNumFiles.

This PR creates a post commit hook, which runs an OptimizeExecutor (from OptimizeTableCommand.scala), which will do the compaction.

Details

We add a post-commit hook in TransactionalWrite, that will check if auto-compaction is needed. If the configs are set such that the write meets the criteria (i.e. AC is enabled, enough small files exist, etc) then partitions that meet the criteria will be reserved, and used to make an OptimizeExecutor targeting those partitions, and with the appropriate config values.

This runs and will compact the files. Partitions are then released for future compactions to consider.

AutoCompact is disabled by default

Configs

There are a number of new configs introduced by this PR, all with prefix spark.databricks.delta.autoCompact. Through a lot of experimentation and user feedback, we found these values to work well across a large range of tables and configurations.

autoCompact.enabled: should auto compaction run? (default: false)
autoCompact.maxFileSize: Target file size produced by auto compaction (default: 128 MB)
autoCompact.minFileSize: Files which are smaller than this threshold (in bytes) will be grouped together and rewritten as larger files by the Auto Compaction. (default: half of maxFileSize)

How was this patch tested?

Unit tests in AutoCompactionSuite

Does this PR introduce any user-facing changes?

Yes, please see the Description

felipepessoto · 2023-12-28T22:29:02Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   *   2. Then we check if the deprecated property `DeltaConfigs.AUTO_OPTIMIZE` is set. If yes, then
+   *      we return [[AutoCompactType.Legacy]] type.
+   *   3. Then we check the table property [[DeltaConfigs.AUTO_COMPACT]].
+   *   4. If none of 1/2/3 are set explicitly, then we return None


It never returns None

We actually do. If you look here in the apply method for AutoCompactType, when we pass it Disabled it returns None

felipepessoto · 2023-12-28T22:29:17Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   *   3. Then we check the table property [[DeltaConfigs.AUTO_COMPACT]].
+   *   4. If none of 1/2/3 are set explicitly, then we return None
+   */
+  def getAutoCompactType(conf: SQLConf, metadata: Metadata): Option[AutoCompactType] = {


Option[AutoCompactType] -> AutoCompactType

See comment about returning None

felipepessoto · 2023-12-28T22:30:48Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   *   1. The highest priority is given to [[DeltaSQLConf.DELTA_AUTO_COMPACT_ENABLED]] config.
+   *   2. Then we check if the deprecated property `DeltaConfigs.AUTO_OPTIMIZE` is set. If yes, then
+   *      we return [[AutoCompactType.Legacy]] type.
+   *   3. Then we check the table property [[DeltaConfigs.AUTO_COMPACT]].


The new flag shouldn't have priority?

felipepessoto · 2023-12-28T22:32:51Z

spark/src/main/scala/org/apache/spark/sql/delta/DeltaConfig.scala

+   * hook to compact the files.
+   *  It can be enabled by setting the property to `true`
+   * Note that the behavior from table property can be overridden by the config:
+   * [[org.apache.spark.sql.delta.sources.DeltaSQLConfEdge.DELTA_AUTO_COMPACT_ENABLED]]


org.apache.spark.sql.delta.sources.DeltaSQLConfEdge I guess this is from your private code

felipepessoto · 2023-12-28T22:34:40Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   * Prioritization:
+   *   1. The highest priority is given to [[DeltaSQLConf.DELTA_AUTO_COMPACT_ENABLED]] config.
+   *   2. Then we check if the deprecated property `DeltaConfigs.AUTO_OPTIMIZE` is set. If yes, then
+   *      we return [[AutoCompactType.Legacy]] type.


I also prefer Legacy, but it is currently named Enabled

felipepessoto · 2023-12-28T22:39:04Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   * 2. MIN_FILE_SIZE is configurable and defaults to MAX_FILE_SIZE / 2 unless overridden.
+   * Note: User can use DELTA_AUTO_COMPACT_MAX_FILE_SIZE to override this value.
+   */
+  case object Enabled extends AutoCompactType {


Should we have two distinct objects? Enabled and Legacy to make it clear?

Legacy is a, well, legacy name, there are only Enabled, or Disabled

felipepessoto · 2023-12-28T22:39:40Z

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

+   * 2. MIN_FILE_SIZE is configurable and defaults to MAX_FILE_SIZE / 2 unless overridden.
+   * Note: User can use DELTA_AUTO_COMPACT_MAX_FILE_SIZE to override this value.
+   */
+  val defaultMaxFileSize: Long = 16 * 1024 * 1024


I don't see any usage of defaultMaxFileSize

Nice catch thanks! A victim of some code shuffling. Fixed to not need this anymore.

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

felipepessoto · 2023-12-28T22:44:02Z

@nicklan, we have another Auto Compact PR: #1156
Do you know what are the differences? Is this implementation the one used by Databricks, meaning it has been tested extensively?

nicklan · 2023-12-29T11:52:16Z

@felipepessoto thanks for the review! Yes, this is based on the implementation used by Databricks and so has been battle tested in many production workloads. We absolutely considered #1156, and are grateful for the contribution. However, due to our extensive experience with this implementation we felt this was the best path forward.

(I'm OOO until early Jan, but will reply as possible to further comments)

jaceklaskowski · 2024-01-04T21:39:32Z

spark/src/main/scala/org/apache/spark/sql/delta/DeltaOperations.scala

@@ -471,6 +471,8 @@ object DeltaOperations {

  sealed abstract class OptimizeOrReorg(override val name: String, predicates: Seq[Expression])
    extends OperationWithPredicates(name, predicates)
+  /** parameter key to indicate whether it's an Auto Compaction */


to indicate Auto Compaction

re #2414 (comment), perhaps we can address post merge

jaceklaskowski · 2024-01-04T21:40:32Z

spark/src/main/scala/org/apache/spark/sql/delta/DeltaOperations.scala

@@ -482,10 +484,12 @@ object DeltaOperations {
  /** Recorded when optimizing the table. */
  case class Optimize(
      predicate: Seq[Expression],
-      zOrderBy: Seq[String] = Seq.empty
+      zOrderBy: Seq[String] = Seq.empty,
+      auto: Boolean = false


Think autoCompact would be more meaningful

re #2414 (comment), perhaps we can address post merge

jaceklaskowski · 2024-01-04T21:41:32Z

spark/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

@@ -350,6 +350,11 @@ trait OptimisticTransactionImpl extends TransactionalWrite
    checkDeletionVectorFilesHaveWideBounds = false
  }

+  /** The set of distinct partitions that contain added files by current transaction. */
+  protected[delta] var partitionsAddedToOpt: Option[mutable.HashSet[Map[String, String]]] = None


Remove Opt suffix (it's Scala - a type-heavy language after all 😉 )

We use this fairly consistently:

https://github.com/delta-io/delta/blob/master/spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala#L62

https://github.com/delta-io/delta/blob/master/spark/src/main/scala/org/apache/spark/sql/delta/MetadataCleanup.scala#L63

https://github.com/delta-io/delta/blob/master/spark/src/main/scala/org/apache/spark/sql/delta/MetadataCleanup.scala#L63

We could propose something here, more generally, but for now I'll leave it as is

jaceklaskowski · 2024-01-04T21:42:37Z

spark/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

@@ -860,6 +865,46 @@ trait OptimisticTransactionImpl extends TransactionalWrite
    }
  }

+  def reportAutoCompactStatsError(e: Throwable): Unit = {


Gotcha! That proves my point about calling the param autoCompact 😉

jaceklaskowski · 2024-01-04T21:44:00Z

spark/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

+    if (numAdd == 0 && numRemove == 0) return
+    val collector = createAutoCompactStatsCollector()
+    if (collector.isInstanceOf[DisabledAutoCompactPartitionStatsCollector]) return
+    AutoCompactPartitionStats.instance(spark)


Is instance needed?! Why not use apply?

see https://github.com/delta-io/delta/pull/2414/files#r1446659873

jaceklaskowski · 2024-01-04T21:44:23Z

spark/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

+   * A subclass of AutoCompactPartitionStatsCollector that's to be used if the config to collect
+   * auto compaction stats is turned off. This subclass intentionally does nothing.
+   */
+  class DisabledAutoCompactPartitionStatsCollector extends AutoCompactPartitionStatsCollector{


A whitespace before closing {

jaceklaskowski · 2024-01-04T21:54:21Z

spark/src/main/scala/org/apache/spark/sql/delta/stats/AutoCompactPartitionStats.scala

@@ -0,0 +1,376 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.


jaceklaskowski · 2024-01-04T21:55:38Z

spark/src/main/scala/org/apache/spark/sql/delta/stats/AutoCompactPartitionStats.scala

+class AutoCompactPartitionStats(
+    private var maxNumTablePartitions: Int,
+    private var maxNumPartitions: Int
+) {


Why on a separate line while the other class PartitionStat has them to close the input arguments?! Be consistent 🙏

re #2414 (comment), perhaps we can address post merge

jaceklaskowski · 2024-01-04T21:56:04Z

spark/src/main/scala/org/apache/spark/sql/delta/stats/AutoCompactPartitionStats.scala

+     * @param minNumFiles The minimum number of files this table-partition should have to trigger
+     *                    Auto Compaction in case it has already been compacted once.
+     */
+    def hasSufficientSmallFilesOrHasNotBeenCompacted(minNumFiles: Long): Boolean =


...OrNotCompactedYet?

re #2414 (comment), perhaps we can address post merge

jaceklaskowski · 2024-01-04T21:57:51Z

spark/src/main/scala/org/apache/spark/sql/delta/stats/AutoCompactPartitionStats.scala

+  private var _instance: AutoCompactPartitionStats = null
+
+  /** The thread safe constructor of singleton. */
+  def instance(spark: SparkSession): AutoCompactPartitionStats = {


Why synchronized? Why is there a need for a singleton?

Multiple threads could be committing simultaneously, but we want to keep all the stats together. This gives us a safe way to have a single source of stats across all transactions

jaceklaskowski · 2024-01-04T21:57:58Z

spark/src/test/scala/org/apache/spark/sql/delta/AutoCompactSuite.scala

@@ -0,0 +1,332 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.


aah... this is a global problem for all files. let fix this in a separate PR for all files.

felipepessoto · 2024-01-05T19:22:50Z

This is not in the roadmap. Is it planned to 3.1 release?

vkorukanti

lgtm after addressing all pending review comments.

vkorukanti · 2024-01-05T19:53:15Z

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

+      .booleanConf
+      .createWithDefault(true)
+
+  val DELTA_AUTO_COMPACT_NON_BLIND_APPEND_ENABLED =


test with this property enabled?

Have a follow-up for some other testing, which we will do post merge

vkorukanti · 2024-01-05T19:53:43Z

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

+      .booleanConf
+      .createWithDefault(false)
+
+  val DELTA_AUTO_COMPACT_MAX_NUM_MODIFIED_PARTITIONS =


same test comment here.

This reverts commit 398547a.

tdas · 2024-01-09T22:49:31Z

Overall, this PR looks pretty good to me other than small naming improvements that others have pointed out (@felipepessoto @jaceklaskowski thank you so so much for your reviews). We are nearing the target 3.1 release date of Jan 24, and I would like to cut a release soon with this feature. Unless there is major objections to the core logic of the PR, can we merge this?

felipepessoto · 2024-01-09T23:06:00Z

I only have one concern, my comment:

The new flag shouldn't have priority?

It that right? Is it the same behavior of Databricks? So that could be a reason to keep compatibility.

I don't see any previous usage of that flag, for example for optimized write we ignored the old flag and don't use it at all.

felipepessoto · 2024-01-09T23:07:22Z

And a nit, the AUTO_COMPACT could be placed close to the other autoOptimize settings

nicklan · 2024-01-10T00:03:04Z

It that right? Is it the same behavior of Databricks? So that could be a reason to keep compatibility.

Yeah, this is to maintain compatibility with Databricks, so I think we should keep it as is

nicklan · 2024-01-10T00:04:51Z

And a nit, the AUTO_COMPACT could be placed close to the other autoOptimize settings

Also makes sense, but is trickier than you might expect. We can absolutely look at cleaning this up post merge if possible

felipepessoto · 2024-01-10T22:16:23Z

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

+      .internal()
+      .doc(s"Target file size produced by auto compaction. The default value of this config" +
+        " is 128 MB.")
+      .longConf


Should we use bytesConf to be consistent with the optimized write?

I'd say to do it after 3.1 cut, but before final release, if possible

nicklan changed the title ~~add auto-compact post commit hook~~ [Spark] add auto-compact post commit hook Dec 28, 2023

felipepessoto reviewed Dec 28, 2023

View reviewed changes

jaceklaskowski reviewed Jan 4, 2024

View reviewed changes

vkorukanti added this to the 3.1.0 milestone Jan 5, 2024

vkorukanti approved these changes Jan 5, 2024

View reviewed changes

nicklan added 4 commits January 8, 2024 13:32

add auto-compact post commit hook

391b1d7

fix whitespace at end of line

ae41766

address pr comments

48cb879

cosmetic

e0184d0

nicklan force-pushed the auto-compact-b branch from 352e02b to e0184d0 Compare January 8, 2024 21:32

nicklan added 10 commits January 9, 2024 10:25

minor adjustments

cc5d39e

spacing adjust

ccb2d82

scalastyle

6db09a3

Merge branch 'master' into auto-compact-b

192f4b5

call with ()

398547a

Revert "call with ()"

c974864

This reverts commit 398547a.

add error test and fix arg order

cd04f8c

Merge branch 'master' into auto-compact-b

d880a1f

Address comments

96efa44

add forgotten import

6f7bd9a

put test in correct location

efd4b30

nicklan and others added 4 commits January 9, 2024 16:35

fix for style

5e02b22

Update AutoCompactSuite.scala

da05450

Merge branch 'master' into auto-compact-b

93f94e1

Merge branch 'master' into auto-compact-b

42fe5a9

felipepessoto reviewed Jan 10, 2024

View reviewed changes

vkorukanti closed this in d18a441 Jan 11, 2024

osopardo1 mentioned this pull request Feb 22, 2024

Upgrade to Delta 3.1.0 Qbeast-io/qbeast-spark#275

Closed

7 tasks

		@@ -0,0 +1,376 @@
		/*
		* Copyright (2021) The Delta Lake Project Authors.

		@@ -0,0 +1,332 @@
		/*
		* Copyright (2021) The Delta Lake Project Authors.

[Spark] add auto-compact post commit hook #2414

[Spark] add auto-compact post commit hook #2414

Conversation

nicklan commented Dec 28, 2023

Which Delta project/connector is this regarding?

Description

Details

Configs

How was this patch tested?

Does this PR introduce any user-facing changes?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felipepessoto commented Dec 28, 2023

nicklan commented Dec 29, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felipepessoto commented Jan 5, 2024

vkorukanti left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Jan 9, 2024

felipepessoto commented Jan 9, 2024

felipepessoto commented Jan 9, 2024

nicklan commented Jan 10, 2024

nicklan commented Jan 10, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nicklan commented Dec 29, 2023 •

edited

Loading