Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Support auto-compaction for Delta tables on [databricks] #7889

Merged
merged 18 commits into from
Mar 27, 2023
Merged

Support auto-compaction for Delta tables on [databricks] #7889

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
e1520a5
Rebased auto-compaction.
mythrocks Mar 10, 2023
fbf60e7
Use GpuOptimisticTransaction for auto-optimized write.
mythrocks Mar 10, 2023
72fd27d
Support for Databricks 10.4.
mythrocks Mar 14, 2023
9923a5b
Test for partitioned case.
mythrocks Mar 14, 2023
176f4a6
Fixed import order.
mythrocks Mar 14, 2023
155eedb
Fix test for 10.4.
mythrocks Mar 15, 2023
a2b2a2c
Review comment: Use System.currentTimeMillis.
mythrocks Mar 15, 2023
5979720
Test for disabled auto-compact.
mythrocks Mar 15, 2023
c5c11a6
Added test for min-num-files.
mythrocks Mar 15, 2023
36d32ef
Fix skipping tests on non-Databricks systems.
mythrocks Mar 16, 2023
4424dd5
Disable GPU-fallback test for autoCompact.
mythrocks Mar 17, 2023
147c2ff
Review updates:
mythrocks Mar 22, 2023
bc2a68d
Added test for auto-compact default.
mythrocks Mar 23, 2023
bf2e848
Added clarifying code comment.
mythrocks Mar 23, 2023
61e8e59
Updated delta-lake-support docs for autoCompact.
mythrocks Mar 23, 2023
35dc58c
Fixed foul-up in auto-compaction documentation.
mythrocks Mar 23, 2023
9456b2c
Merge remote-tracking branch 'origin/branch-23.04' into rebased-auto-…
mythrocks Mar 23, 2023
74e6afe
Removed confusing distinction of SQL writes vs Scala/Python.
mythrocks Mar 24, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion ...21db/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuDoAutoCompaction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,13 +32,16 @@ import org.apache.spark.sql.SparkSession
object GpuDoAutoCompaction extends PostCommitHook
with DeltaLogging
with Serializable {
override val name: String = "GpuDoAutoCompaction"
override val name: String = "Triggers compaction if necessary"

override def run(spark: SparkSession,
txn: OptimisticTransactionImpl,
committedActions: Seq[Action]): Unit = {
val gpuTxn = txn.asInstanceOf[GpuOptimisticTransaction]
val newTxn = new GpuDeltaLog(gpuTxn.deltaLog, gpuTxn.rapidsConf).startTransaction()
// Note: The Databricks AutoCompact PostCommitHook cannot be used here
// (with a GpuOptimisticTransaction). It does not appear to use OptimisticTransaction.writeFiles
// to write the compacted file.
mythrocks marked this conversation as resolved.
Show resolved Hide resolved
new GpuOptimizeExecutor(spark, newTxn, Seq.empty, Seq.empty, committedActions).optimize()
jlowe marked this conversation as resolved.
Show resolved Hide resolved
}

Expand Down
2 changes: 1 addition & 1 deletion ...src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuOptimisticTransaction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,7 @@ class GpuOptimisticTransaction(
.getConf[String](DeltaSQLConf.DELTA_AUTO_COMPACT_ENABLED)
.getOrElse {
DeltaConfigs.AUTO_COMPACT.fromMetaData(metadata)
"false" // TODO: Fix getting this from DeltaConfigs.AUTO_COMPACT.
.getOrElse("false")
}.toBoolean

if (!isOptimize && autoCompactEnabled && fileActions.nonEmpty) {
Expand Down
24 changes: 12 additions & 12 deletions ...21db/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuOptimizeExecutor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ class GpuOptimizeExecutor(

private val isMultiDimClustering = zOrderByColumns.nonEmpty
private val isAutoCompact = prevCommitActions.nonEmpty
private val optimizeType = OptimizeType(isMultiDimClustering, isAutoCompact)
private val optimizeType = GpuOptimizeType(isMultiDimClustering, isAutoCompact)

def optimize(): Seq[Row] = {
recordDeltaOperation(txn.deltaLog, "delta.optimize") {
Expand Down Expand Up @@ -206,7 +206,7 @@ class GpuOptimizeExecutor(
// zOrderByColumns)
throw new UnsupportedOperationException("MultiDimClustering not supported on compaction")
} else {
// Re-partition is not available in Databricks 11.3 (spark321db)
// Re-partition is not available in Databricks 10.4 (spark321db)
input.coalesce(numPartitions = 1)
}

Expand All @@ -231,9 +231,9 @@ class GpuOptimizeExecutor(
updates
}

type PartitionedBin = (Map[String, String], Seq[AddFile])
private type PartitionedBin = (Map[String, String], Seq[AddFile])

trait OptimizeType {
private trait GpuOptimizeType {
def minNumFiles: Long

def maxFileSize: Long =
Expand All @@ -244,7 +244,7 @@ class GpuOptimizeExecutor(
def targetBins(jobs: Seq[PartitionedBin]): Seq[PartitionedBin] = jobs
}

case class Compaction() extends OptimizeType {
private case class GpuCompaction() extends GpuOptimizeType {
def minNumFiles: Long = 2

def targetFiles: (Seq[AddFile], Seq[AddFile]) = {
Expand All @@ -257,7 +257,7 @@ class GpuOptimizeExecutor(
}
}

case class MultiDimOrdering() extends OptimizeType {
private case class GpuMultiDimOrdering() extends GpuOptimizeType {
def minNumFiles: Long = 1

def targetFiles: (Seq[AddFile], Seq[AddFile]) = {
Expand All @@ -267,7 +267,7 @@ class GpuOptimizeExecutor(
}
}

case class AutoCompaction() extends OptimizeType {
private case class GpuAutoCompaction() extends GpuOptimizeType {
def minNumFiles: Long = {
val minNumFiles =
sparkSession.sessionState.conf.getConf(DeltaSQLConf.DELTA_AUTO_COMPACT_MIN_NUM_FILES)
Expand Down Expand Up @@ -315,15 +315,15 @@ class GpuOptimizeExecutor(
}
}

object OptimizeType {
private object GpuOptimizeType {

def apply(isMultiDimClustering: Boolean, isAutoCompact: Boolean): OptimizeType = {
def apply(isMultiDimClustering: Boolean, isAutoCompact: Boolean): GpuOptimizeType = {
if (isMultiDimClustering) {
MultiDimOrdering()
GpuMultiDimOrdering()
} else if (isAutoCompact) {
AutoCompaction()
GpuAutoCompaction()
} else {
Compaction()
GpuCompaction()
}
}
}
Expand Down
5 changes: 4 additions & 1 deletion ...30db/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuDoAutoCompaction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import org.apache.spark.sql.SparkSession
object GpuDoAutoCompaction extends PostCommitHook
with DeltaLogging
with Serializable {
override val name: String = "GpuDoAutoCompaction"
override val name: String = "Triggers compaction if necessary"

override def run(spark: SparkSession,
txn: OptimisticTransactionImpl,
Expand All @@ -41,6 +41,9 @@ object GpuDoAutoCompaction extends PostCommitHook
committedActions: Seq[Action]): Unit = {
val gpuTxn = txn.asInstanceOf[GpuOptimisticTransaction]
val newTxn = new GpuDeltaLog(gpuTxn.deltaLog, gpuTxn.rapidsConf).startTransaction()
// Note: The Databricks AutoCompact PostCommitHook cannot be used here
// (with a GpuOptimisticTransaction). It does not appear to use OptimisticTransaction.writeFiles
// to write the compacted file.
mythrocks marked this conversation as resolved.
Show resolved Hide resolved
new GpuOptimizeExecutor(spark, newTxn, Seq.empty, Seq.empty, committedActions).optimize()
}

Expand Down
4 changes: 2 additions & 2 deletions ...src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuOptimisticTransaction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,8 +288,8 @@ class GpuOptimisticTransaction(
spark.sessionState.conf
.getConf[String](DeltaSQLConf.DELTA_AUTO_COMPACT_ENABLED)
.getOrElse {
// DeltaConfigs.AUTO_COMPACT.fromMetaData(metadata)
"false" // TODO: Fix getting this from DeltaConfigs.AUTO_COMPACT.
DeltaConfigs.AUTO_COMPACT.fromMetaData(metadata)
.getOrElse("false")
}.toBoolean

if (!isOptimize && autoCompactEnabled && fileActions.nonEmpty) {
Expand Down
22 changes: 11 additions & 11 deletions ...30db/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuOptimizeExecutor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ class GpuOptimizeExecutor(

private val isMultiDimClustering = zOrderByColumns.nonEmpty
private val isAutoCompact = prevCommitActions.nonEmpty
private val optimizeType = OptimizeType(isMultiDimClustering, isAutoCompact)
private val optimizeType = GpuOptimizeType(isMultiDimClustering, isAutoCompact)

def optimize(): Seq[Row] = {
recordDeltaOperation(txn.deltaLog, "delta.optimize") {
Expand Down Expand Up @@ -235,9 +235,9 @@ class GpuOptimizeExecutor(
updates
}

type PartitionedBin = (Map[String, String], Seq[AddFile])
private type PartitionedBin = (Map[String, String], Seq[AddFile])

trait OptimizeType {
private trait GpuOptimizeType {
def minNumFiles: Long

def maxFileSize: Long =
Expand All @@ -248,7 +248,7 @@ class GpuOptimizeExecutor(
def targetBins(jobs: Seq[PartitionedBin]): Seq[PartitionedBin] = jobs
}

case class Compaction() extends OptimizeType {
private case class GpuCompaction() extends GpuOptimizeType {
def minNumFiles: Long = 2

def targetFiles: (Seq[AddFile], Seq[AddFile]) = {
Expand All @@ -261,7 +261,7 @@ class GpuOptimizeExecutor(
}
}

case class MultiDimOrdering() extends OptimizeType {
private case class GpuMultiDimOrdering() extends GpuOptimizeType {
def minNumFiles: Long = 1

def targetFiles: (Seq[AddFile], Seq[AddFile]) = {
Expand All @@ -271,7 +271,7 @@ class GpuOptimizeExecutor(
}
}

case class AutoCompaction() extends OptimizeType {
private case class GpuAutoCompaction() extends GpuOptimizeType {
def minNumFiles: Long = {
val minNumFiles =
sparkSession.sessionState.conf.getConf(DeltaSQLConf.DELTA_AUTO_COMPACT_MIN_NUM_FILES)
Expand Down Expand Up @@ -319,15 +319,15 @@ class GpuOptimizeExecutor(
}
}

object OptimizeType {
private object GpuOptimizeType {

def apply(isMultiDimClustering: Boolean, isAutoCompact: Boolean): OptimizeType = {
def apply(isMultiDimClustering: Boolean, isAutoCompact: Boolean): GpuOptimizeType = {
if (isMultiDimClustering) {
MultiDimOrdering()
GpuMultiDimOrdering()
} else if (isAutoCompact) {
AutoCompaction()
GpuAutoCompaction()
} else {
Compaction()
GpuCompaction()
}
}
}
Expand Down