From 4fcdc8d877b1f5fc4fa39932033362cd50f82b84 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 6 Oct 2026 13:51:25 -0700 Subject: [PATCH] [Spark] Lightweight compaction for clustered tables Add an opt-in lightweight tier to OPTIMIZE and auto compaction on clustered tables. While the table has some unclustered data, but less than spark.databricks.delta.optimize.clustering.lightweight.maxUnclusteredBytes, OPTIMIZE compacts only the small unclustered files instead of clustering: clustered files and partial Z-cubes are not rewritten, and rows are not sorted. Candidate files are ordered by the per-file min/max statistics of a single clustering column before they are packed into bins, so each output file covers a narrow range of that column. The column is configurable, or picked as the clustering column whose files overlap least. Output files stay unclustered and are clustered once enough unclustered data has accumulated. OPTIMIZE FULL always clusters. Configs: - optimize.clustering.lightweight.enabled (default false) - optimize.clustering.lightweight.maxUnclusteredBytes (default 10GB) - optimize.clustering.lightweight.column (default: auto-pick) - optimize.clustering.lightweight.maxRelativeFileRange (default 0.5) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/src/content/docs/delta-clustering.mdx | 8 + .../resources/error/delta-error-classes.json | 6 + .../apache/spark/sql/delta/DeltaErrors.scala | 11 + .../delta/commands/OptimizeTableCommand.scala | 13 +- .../commands/OptimizeTableStrategy.scala | 106 +++++- .../ClusteringColumnFileOrdering.scala | 161 +++++++++ .../sql/delta/sources/DeltaSQLConf.scala | 42 +++ .../LightweightClusteringSuite.scala | 307 ++++++++++++++++++ 8 files changed, 645 insertions(+), 9 deletions(-) create mode 100644 spark/src/main/scala/org/apache/spark/sql/delta/commands/optimize/ClusteringColumnFileOrdering.scala create mode 100644 spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/LightweightClusteringSuite.scala diff --git a/docs/src/content/docs/delta-clustering.mdx b/docs/src/content/docs/delta-clustering.mdx index ac239475341..56e087bb21b 100644 --- a/docs/src/content/docs/delta-clustering.mdx +++ b/docs/src/content/docs/delta-clustering.mdx @@ -146,6 +146,14 @@ In Delta Lake 3.3 and above, you can force reclustering of all records in a tabl Run `OPTIMIZE FULL` when you change clustering columns. If you have previously run `OPTIMIZE FULL` and there has been no change to clustering columns, `OPTIMIZE FULL` runs the same as `OPTIMIZE`. Always use `OPTIMIZE FULL` to ensure that data layout reflects the current clustering columns. +### Lightweight compaction + +Each `OPTIMIZE` clusters the new data together with previously clustered data that has not yet reached the target size, and sorts all of it. On tables that receive small amounts of data between frequent `OPTIMIZE` runs, the same data can be rewritten many times. To reduce this cost, set `spark.databricks.delta.optimize.clustering.lightweight.enabled` to `true`. While the table has less unclustered data than `spark.databricks.delta.optimize.clustering.lightweight.maxUnclusteredBytes` (10 GB by default), `OPTIMIZE` and auto compaction then only compact small unclustered files, and leave clustered files untouched: + +- Files are grouped by the minimum and maximum values of one clustering column, so each compacted file covers a narrow range of that column. Rows are not sorted. +- Set `spark.databricks.delta.optimize.clustering.lightweight.column` to choose the column. Otherwise, the clustering column whose files cover the narrowest part of its value range is used, such as a date column on a table written in daily batches. If the files of every clustering column cover more than `spark.databricks.delta.optimize.clustering.lightweight.maxRelativeFileRange` (0.5 by default) of its value range on average, files are grouped by size. +- The compacted files remain unclustered. Once the table has accumulated `maxUnclusteredBytes` of unclustered data, `OPTIMIZE` clusters it. `OPTIMIZE FULL` always clusters. + ## Read data from a clustered table You can read data in a clustered table using any Delta Lake client. For best query results, include clustering columns in your query filters, as in the following example: diff --git a/spark/src/main/resources/error/delta-error-classes.json b/spark/src/main/resources/error/delta-error-classes.json index 22302558785..bd30c55bd1d 100644 --- a/spark/src/main/resources/error/delta-error-classes.json +++ b/spark/src/main/resources/error/delta-error-classes.json @@ -1995,6 +1995,12 @@ ], "sqlState" : "25000" }, + "DELTA_INVALID_LIGHTWEIGHT_CLUSTERING_COLUMN" : { + "message" : [ + "Column '' in '' is not a clustering column of the table. Clustering columns: ." + ], + "sqlState" : "42703" + }, "DELTA_INVALID_LOGSTORE_CONF" : { "message" : [ "(``) and (``) cannot be set at the same time. Please set only one group of them." diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala index 17080e7ed66..5fd9504add6 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala @@ -4156,6 +4156,17 @@ trait DeltaErrorsBase ) } + def invalidLightweightClusteringColumn( + column: String, clusteringColumns: Seq[String]): Throwable = { + new DeltaAnalysisException( + errorClass = "DELTA_INVALID_LIGHTWEIGHT_CLUSTERING_COLUMN", + messageParameters = Array( + column, + DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_COLUMN.key, + clusteringColumns.mkString(", ")) + ) + } + def clusteringColumnUnsupportedDataTypes(clusteringColumnsWithDataTypes: String): Throwable = { new DeltaAnalysisException( errorClass = "DELTA_CLUSTERING_COLUMNS_DATATYPE_NOT_SUPPORTED", diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala index 4f186379eab..c33a2c4f363 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala @@ -284,6 +284,9 @@ class OptimizeExecutor( private val isClusteredTable = ClusteredTableUtils.isSupported(snapshot.protocol) + private val isLightweightClustering = + optimizeStrategy.isInstanceOf[LightweightClusteringStrategy] + private val isMultiDimClustering = optimizeStrategy.isInstanceOf[ClusteringStrategy] || optimizeStrategy.isInstanceOf[ZOrderStrategy] @@ -314,12 +317,12 @@ class OptimizeExecutor( // transactions later val candidateFiles = snapshot.filesForScan(partitionPredicate, keepNumRecords = true).files - val filesToProcess = optimizeContext.reorg match { + val filesToProcess = optimizeStrategy.prepareFiles(optimizeContext.reorg match { case Some(reorgOperation) => reorgOperation.filterFilesToReorg(sparkSession, snapshot, candidateFiles) case None => filterCandidateFileList(minFileSize, maxDeletedRowsRatio, candidateFiles) - } + }) // Group files by their normalized (typed) partition values so that logically equivalent // but differently formatted values (e.g. timestamp variants) end up in the same group. val partitionsToCompact = filesToProcess @@ -608,7 +611,11 @@ class OptimizeExecutor( predicate = partitionPredicate, zOrderBy = zOrderByColumns, auto = isAutoCompact, - clusterBy = if (isClusteredTable) Option(clusteringColumns).filter(_.nonEmpty) else None, + clusterBy = if (isClusteredTable && !isLightweightClustering) { + Option(clusteringColumns).filter(_.nonEmpty) + } else { + None + }, isFull = optimizeContext.isFull) } } diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableStrategy.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableStrategy.scala index 592d658cbce..c0ddea4756f 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableStrategy.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableStrategy.scala @@ -21,13 +21,15 @@ import org.apache.spark.sql.delta.skipping.clustering.ZCube import org.apache.spark.sql.delta.{DeltaConfigs, DeltaErrors, OptimisticTransaction, Snapshot} import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor, FileAction, RemoveFile} import org.apache.spark.sql.delta.commands.OptimizeTableStrategy.DummyBinInfo -import org.apache.spark.sql.delta.commands.optimize.{AddFileWithNumRecords, DeletionVectorStats, OptimizeStats, ZOrderFileStats, ZOrderStats} +import org.apache.spark.sql.delta.commands.optimize.{AddFileWithNumRecords, ClusteringColumnFileOrdering, DeletionVectorStats, OptimizeStats, ZOrderFileStats, ZOrderStats} +import org.apache.spark.sql.delta.metering.DeltaLogging import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.sources.DeltaSQLConf.{DELTA_OPTIMIZE_CLUSTERING_MIN_CUBE_SIZE, DELTA_OPTIMIZE_CLUSTERING_TARGET_CUBE_SIZE} import org.apache.spark.sql.delta.zorder.ZCubeInfo import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.functions.{coalesce, col, lit, sum} object OptimizeTableMode extends Enumeration { type OptimizeTableMode = Value @@ -46,6 +48,14 @@ trait OptimizeTableStrategy { */ def maxBinSize: Long + /** + * Utility method to prepare all files selected for optimization, before they are grouped by + * partition and passed to [[prepareFilesPerPartition]]. + * + * @return Prepared files for the subsequent optimization. + */ + def prepareFiles(inputFiles: Seq[AddFile]): Seq[AddFile] = inputFiles + /** * Utility method to prepare files in a partition for optimization. * @@ -93,7 +103,7 @@ trait OptimizeTableStrategy { bins: Seq[Bin]): Unit } -object OptimizeTableStrategy { +object OptimizeTableStrategy extends DeltaLogging { // A trait representing the context for a Bin. sealed trait BinInfo @@ -109,16 +119,47 @@ object OptimizeTableStrategy { optimizeContext: DeltaOptimizeContext, zOrderBy: Seq[String]): OptimizeTableStrategy = getMode(snapshot, zOrderBy) match { case OptimizeTableMode.CLUSTERING => - ClusteringStrategy( - sparkSession, - ClusteringColumnInfo.extractLogicalNames(snapshot), - optimizeContext) + val clusteringColumns = ClusteringColumnInfo.extractLogicalNames(snapshot) + if (useLightweightClustering(sparkSession, snapshot, optimizeContext)) { + LightweightClusteringStrategy(sparkSession, snapshot, clusteringColumns, optimizeContext) + } else { + ClusteringStrategy(sparkSession, clusteringColumns, optimizeContext) + } case OptimizeTableMode.ZORDER => ZOrderStrategy(sparkSession, zOrderBy) case OptimizeTableMode.COMPACTION => CompactionStrategy(sparkSession, optimizeContext) case other => throw new UnsupportedOperationException(s"Unsupported mode $other") } + /** + * Whether OPTIMIZE on a clustered table should run lightweight compaction instead of + * clustering: the table has some unclustered data, but less than + * [[DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_UNCLUSTERED_BYTES]]. Without any + * unclustered data, clustering still runs so that partial Z-cubes can be merged. + */ + private def useLightweightClustering( + sparkSession: SparkSession, + snapshot: Snapshot, + optimizeContext: DeltaOptimizeContext): Boolean = { + val conf = sparkSession.sessionState.conf + if (!conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_ENABLED) || + optimizeContext.isFull || optimizeContext.reorg.nonEmpty) { + return false + } + val unclusteredBytes = snapshot.allFiles + .where(col("clusteringProvider").isNull) + .select(coalesce(sum(col("size")), lit(0L))) + .head() + .getLong(0) + val threshold = + conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_UNCLUSTERED_BYTES) + val useLightweight = unclusteredBytes > 0 && unclusteredBytes < threshold + logInfo(s"Clustered table has $unclusteredBytes bytes of unclustered data, threshold for " + + s"clustering is $threshold bytes: " + + (if (useLightweight) "running lightweight compaction" else "clustering")) + useLightweight + } + private def getMode(snapshot: Snapshot, zOrderBy: Seq[String]): OptimizeTableMode.Value = { val isClusteredTable = ClusteredTableUtils.isSupported(snapshot.protocol) val hasClusteringColumns = ClusteringColumnInfo.extractLogicalNames(snapshot).nonEmpty @@ -157,6 +198,59 @@ case class CompactionStrategy( bins: Seq[Bin]): Unit = {} } +/** + * Implements lightweight compaction for clustered tables, used instead of [[ClusteringStrategy]] + * while the table has little unclustered data (see + * [[DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_ENABLED]]). + * + * Clustering rewrites the new data together with all partial Z-cubes and sorts every row, so + * running it often on small amounts of new data rewrites the same data many times. Instead, this + * strategy only compacts small unclustered files, like [[CompactionStrategy]], and leaves + * clustered files untouched. The files are grouped by the min/max statistics of one clustering + * column (see [[ClusteringColumnFileOrdering]]) so that each output file covers a narrow range of + * that column, and rows are not sorted. The output files stay unclustered, so they are clustered + * once the table has accumulated enough unclustered data. + */ +case class LightweightClusteringStrategy( + override val sparkSession: SparkSession, + snapshot: Snapshot, + clusteringColumns: Seq[String], + optimizeContext: DeltaOptimizeContext) extends OptimizeTableStrategy with DeltaLogging { + + override val optimizeTableMode: OptimizeTableMode.Value = OptimizeTableMode.COMPACTION + + override val maxBinSize: Long = { + optimizeContext.maxFileSize.getOrElse( + sparkSession.sessionState.conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_MAX_FILE_SIZE)) + } + + override def curve: String = { + throw new UnsupportedOperationException("Lightweight compaction doesn't sort rows.") + } + + override def prepareFiles(inputFiles: Seq[AddFile]): Seq[AddFile] = { + val conf = sparkSession.sessionState.conf + val (orderedFiles, column) = ClusteringColumnFileOrdering.order( + sparkSession, + snapshot, + inputFiles.filter(_.clusteringProvider.isEmpty), + clusteringColumns, + conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_COLUMN), + conf.getConf(DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_RELATIVE_FILE_RANGE)) + logInfo(s"Lightweight compaction of ${orderedFiles.size} unclustered files, grouped by " + + column.map(c => s"the statistics of column $c").getOrElse("size")) + orderedFiles + } + + // The files were already ordered by prepareFiles. Clustered tables are unpartitioned. + override def prepareFilesPerPartition(inputFiles: Seq[AddFile]): Seq[AddFile] = inputFiles + + override def updateOptimizeStats( + optimizeStats: OptimizeStats, + removedFiles: Seq[RemoveFile], + bins: Seq[Bin]): Unit = {} +} + /** Implements ZOrder strategy */ case class ZOrderStrategy( override val sparkSession: SparkSession, diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/optimize/ClusteringColumnFileOrdering.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/optimize/ClusteringColumnFileOrdering.scala new file mode 100644 index 00000000000..8e9c66f26da --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/optimize/ClusteringColumnFileOrdering.scala @@ -0,0 +1,161 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta.commands.optimize + +import java.util.{Arrays, Comparator} + +import org.apache.spark.sql.delta.{DeltaErrors, Snapshot} +import org.apache.spark.sql.delta.actions.AddFile +import org.apache.spark.sql.delta.stats.DeltaStatistics.{MAX, MIN} + +import org.apache.spark.sql.{Column, Encoders, SparkSession} +import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute +import org.apache.spark.sql.functions.col + +/** + * Orders the files of a clustered table for lightweight compaction by the per-file min/max + * statistics of a single clustering column. + * + * Compaction packs files into bins sequentially in the given order and writes each bin out as one + * file without sorting rows. Ordering the files by the (min, max) values of a column makes each + * bin cover a contiguous range of that column, so data skipping on the column keeps working after + * compaction. Only one column is used: grouping whole files cannot keep several columns narrow at + * once, and clustering takes care of all clustering columns later. + */ +object ClusteringColumnFileOrdering { + + /** The min and max values of a column in a file. Both are null without statistics. */ + case class ValueRange(min: Any, max: Any) { + def isKnown: Boolean = min != null && max != null + } + + /** + * Returns the files ordered for lightweight compaction, and the clustering column used to + * order them. If `column` is not set, the clustering column whose files cover the narrowest + * part of its value range is used, provided that [[relativeFileRange]] is at most + * `maxRelativeFileRange`. If no column qualifies, the files are ordered by size. + */ + def order( + spark: SparkSession, + snapshot: Snapshot, + files: Seq[AddFile], + clusteringColumns: Seq[String], + column: Option[String], + maxRelativeFileRange: Double): (Seq[AddFile], Option[String]) = { + val columns = column match { + case Some(c) => Seq(resolve(c, clusteringColumns)) + case None => clusteringColumns + } + if (files.size < 2 || columns.isEmpty) return (files.sortBy(_.size), None) + + val ranges = loadRanges(spark, snapshot, files, columns.map(nameParts)) + def rangesOf(i: Int): Seq[ValueRange] = + files.map(f => ranges.get(f.path).map(_(i)).getOrElse(ValueRange(null, null))) + + val chosen = if (column.isDefined) { + Some(0) + } else { + columns.indices + .flatMap(i => relativeFileRange(rangesOf(i)).map(i -> _)) + .filter(_._2 <= maxRelativeFileRange) + .sortBy(_._2) + .headOption + .map(_._1) + } + chosen match { + case Some(i) => (sort(files.zip(rangesOf(i))), Some(columns(i))) + case None => (files.sortBy(_.size), None) + } + } + + /** + * Returns how much of the value range of a column a file covers on average, from 0 when no two + * files overlap to 1 when every file covers all values. Values are replaced by their rank among + * the distinct min/max values of all files, so the measure works for any orderable type and + * reflects how much files overlap rather than how values are distributed. Returns None if fewer + * than two files have statistics or all values are equal. + */ + private[delta] def relativeFileRange(ranges: Seq[ValueRange]): Option[Double] = { + val known = ranges.filter(_.isKnown) + if (known.size < 2) return None + val values = known.flatMap(r => Seq(r.min, r.max)).map(_.asInstanceOf[AnyRef]).toArray + Arrays.sort(values, valueComparator) + val distinct = values.foldLeft(Vector.empty[AnyRef]) { (acc, v) => + if (acc.nonEmpty && valueComparator.compare(acc.last, v) == 0) acc else acc :+ v + }.toArray + if (distinct.length < 2) return None + def rank(v: Any): Int = Arrays.binarySearch(distinct, v.asInstanceOf[AnyRef], valueComparator) + val totalWidth = known.map(r => (rank(r.max) - rank(r.min)).toLong).sum + Some(totalWidth.toDouble / known.size / (distinct.length - 1)) + } + + /** Orders files by (min, max), with files without statistics last, ordered by size. */ + private def sort(files: Seq[(AddFile, ValueRange)]): Seq[AddFile] = { + files.sortWith { case ((f1, r1), (f2, r2)) => + val c = compareRanges(r1, r2) + if (c != 0) c < 0 else if (f1.size != f2.size) f1.size < f2.size else f1.path < f2.path + }.map(_._1) + } + + private[delta] def compareRanges(r1: ValueRange, r2: ValueRange): Int = { + if (!r1.isKnown || !r2.isKnown) return java.lang.Boolean.compare(!r1.isKnown, !r2.isKnown) + val c = valueComparator.compare(r1.min.asInstanceOf[AnyRef], r2.min.asInstanceOf[AnyRef]) + if (c != 0) c else { + valueComparator.compare(r1.max.asInstanceOf[AnyRef], r2.max.asInstanceOf[AnyRef]) + } + } + + private val valueComparator: Comparator[AnyRef] = new Comparator[AnyRef] { + override def compare(a: AnyRef, b: AnyRef): Int = + a.asInstanceOf[Comparable[AnyRef]].compareTo(b) + } + + /** Loads the min/max values of the given columns, by file path. */ + private def loadRanges( + spark: SparkSession, + snapshot: Snapshot, + files: Seq[AddFile], + columns: Seq[Seq[String]]): Map[String, IndexedSeq[ValueRange]] = { + val statsColumns: Seq[Column] = columns.flatMap { parts => + Seq(MIN, MAX).map(snapshot.getStatsColumnOrNullLiteral(_, parts.reverse)) + } + val paths = spark.createDataset(files.map(_.path).distinct)(Encoders.STRING).toDF("path") + snapshot.withStats + .join(paths, "path") + .select(col("path") +: statsColumns: _*) + .collect() + .map { row => + row.getString(0) -> + columns.indices.map(i => ValueRange(row.get(1 + 2 * i), row.get(2 + 2 * i))) + } + .toMap + } + + private def nameParts(column: String): Seq[String] = + UnresolvedAttribute.quotedString(column).nameParts + + private def resolve(column: String, clusteringColumns: Seq[String]): String = { + val parts = nameParts(column) + def matches(c: String): Boolean = { + val other = nameParts(c) + other.size == parts.size && other.zip(parts).forall { case (a, b) => a.equalsIgnoreCase(b) } + } + clusteringColumns.find(matches).getOrElse { + throw DeltaErrors.invalidLightweightClusteringColumn(column, clusteringColumns) + } + } +} diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala index 2a4d9453584..3bd10957efc 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala @@ -3439,6 +3439,48 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .checkValue(_ >= 0, "the target must be >= 0") .createWithDefault((DELTA_OPTIMIZE_CLUSTERING_MIN_CUBE_SIZE.defaultValue.get * 1.5).toLong) + val DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_ENABLED = + buildConf("optimize.clustering.lightweight.enabled") + .doc( + "When true, OPTIMIZE and auto compaction on a clustered table only compact the small " + + "unclustered files, without clustering them, while the table has less unclustered data " + + "than optimize.clustering.lightweight.maxUnclusteredBytes. The files are grouped by " + + "the per-file min/max statistics of one clustering column, so each output file covers " + + "a narrow range of that column. Existing Z-cubes are not rewritten. OPTIMIZE FULL " + + "always clusters.") + .booleanConf + .createWithDefault(false) + + val DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_UNCLUSTERED_BYTES = + buildConf("optimize.clustering.lightweight.maxUnclusteredBytes") + .doc( + "Amount of unclustered data, in bytes, at which OPTIMIZE on a clustered table clusters " + + "again instead of running lightweight compaction. Only used when " + + "optimize.clustering.lightweight.enabled is true.") + .bytesConf(ByteUnit.BYTE) + .checkValue(_ >= 0, "the threshold must be >= 0") + .createWithDefault(10 * DELTA_OPTIMIZE_MAX_FILE_SIZE.defaultValue.get) + + val DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_COLUMN = + buildConf("optimize.clustering.lightweight.column") + .doc( + "Clustering column whose per-file min/max statistics lightweight compaction uses to " + + "group files. If not set, the clustering column whose files cover the narrowest part " + + "of its value range is used.") + .stringConf + .createOptional + + val DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_RELATIVE_FILE_RANGE = + buildConf("optimize.clustering.lightweight.maxRelativeFileRange") + .doc( + "When optimize.clustering.lightweight.column is not set, a clustering column is only " + + "picked to group files if the average range of its values in a file, relative to the " + + "range of values in all files to compact, is at most this ratio. If no column " + + "qualifies, files are grouped by size.") + .doubleConf + .checkValue(r => r >= 0 && r <= 1, "the ratio must be between 0 and 1") + .createWithDefault(0.5) + ////////////////// // Clustered Table ////////////////// diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/LightweightClusteringSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/LightweightClusteringSuite.scala new file mode 100644 index 00000000000..a673a8ffac7 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/LightweightClusteringSuite.scala @@ -0,0 +1,307 @@ +/* + * Copyright (2021) The Delta Lake Project Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.delta.skipping.clustering + +import java.sql.Date + +import org.apache.spark.sql.delta.{DeltaAnalysisException, DeltaLog, DeltaOperations} +import org.apache.spark.sql.delta.actions.AddFile +import org.apache.spark.sql.delta.commands.optimize.ClusteringColumnFileOrdering +import org.apache.spark.sql.delta.commands.optimize.ClusteringColumnFileOrdering.ValueRange +import org.apache.spark.sql.delta.sources.DeltaSQLConf +import org.apache.spark.sql.delta.stats.DeltaStatistics.{MAX, MIN} +import org.apache.spark.sql.delta.test.{DeltaSQLCommandTest, DeltaSQLTestUtils} + +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.test.SharedSparkSession + +class LightweightClusteringSuite extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest + with DeltaSQLTestUtils { + + private val enabledKey = DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_ENABLED.key + private val columnKey = DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_COLUMN.key + private val thresholdKey = + DeltaSQLConf.DELTA_OPTIMIZE_CLUSTERING_LIGHTWEIGHT_MAX_UNCLUSTERED_BYTES.key + private val startDate = Date.valueOf("2026-01-01") + + // One file is written per day. The row counts make the size order interleave the days + // (0, 2, 4, 6, 1, 3, 5, 7), so grouping files by size mixes non-adjacent days. + private val rowsPerDay = Seq(0, 4, 1, 5, 2, 6, 3, 7).map(1000 + 50 * _) + + private val groupedByDay = Set((0, 1), (2, 3), (4, 5), (6, 7)) + private val groupedBySize = Set((0, 2), (4, 6), (1, 3), (5, 7)) + + private def createTable( + path: String, + clusterBy: String, + columnMappingMode: String = "none"): Unit = { + sql( + s"""CREATE TABLE delta.`$path` (id BIGINT, d DATE, n STRUCT, s STRING) + |USING delta CLUSTER BY ($clusterBy) + |TBLPROPERTIES ('delta.columnMapping.mode' = '$columnMappingMode')""".stripMargin) + } + + private def appendOneFilePerDay(path: String, days: Seq[Int] = rowsPerDay.indices): Unit = { + days.foreach { day => + spark.range(rowsPerDay(day)) + .selectExpr( + "id", + s"date_add(date'$startDate', $day) AS d", + s"named_struct('d', date_add(date'$startDate', $day)) AS n", + "cast(id AS STRING) AS s") + .coalesce(1) + .write.format("delta").mode("append").save(path) + } + } + + private def files(path: String): Seq[AddFile] = + DeltaLog.forTable(spark, path).update().allFiles.collect().toSeq + + /** Bins hold exactly two of the files written per day: any two fit, no three do. */ + private def maxFileSizeForTwoFileBins(path: String): Long = { + val sizes = files(path).map(_.size).sorted + assert(sizes.take(3).sum > sizes.takeRight(2).sum) + sizes.takeRight(2).sum + } + + /** The (min, max) day of the column `d` of the given files, or of every file in the table. */ + private def dayRanges(path: String, paths: Option[Set[String]] = None): Set[(Int, Int)] = { + val snapshot = DeltaLog.forTable(spark, path).update() + def day(date: Date): Int = + (date.toLocalDate.toEpochDay - startDate.toLocalDate.toEpochDay).toInt + snapshot.withStats + .select( + snapshot.getStatsColumnOrNullLiteral(MIN, Seq("d")), + snapshot.getStatsColumnOrNullLiteral(MAX, Seq("d")), + snapshot.withStats.col("path")) + .collect() + .filter(r => paths.forall(_.contains(r.getString(2)))) + .map(r => (day(r.getDate(0)), day(r.getDate(1)))) + .toSet + } + + private def optimize(path: String, maxFileSize: Long, full: Boolean = false): Unit = { + withSQLConf(DeltaSQLConf.DELTA_OPTIMIZE_MAX_FILE_SIZE.key -> maxFileSize.toString) { + sql(s"OPTIMIZE delta.`$path`${if (full) " FULL" else ""}") + } + } + + private def lastOptimizeClusterBy(path: String): String = { + val lastCommit = DeltaLog.forTable(spark, path).history.getHistory(Some(1)).head + assert(lastCommit.operation === DeltaOperations.OPTIMIZE_OPERATION_NAME) + lastCommit.operationParameters(DeltaOperations.CLUSTERING_PARAMETER_KEY) + } + + private def isClustered(file: AddFile): Boolean = file.clusteringProvider.nonEmpty + + private def checkData(path: String, days: Seq[Int] = rowsPerDay.indices): Unit = { + val expected = days.flatMap { day => + (0 until rowsPerDay(day)).map { id => + val date = Date.valueOf(startDate.toLocalDate.plusDays(day)) + Row(id.toLong, date, Row(date), id.toString) + } + } + checkAnswer(spark.read.format("delta").load(path), expected) + } + + test("OPTIMIZE clusters by default") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path) + optimize(path, maxFileSizeForTwoFileBins(path)) + assert(files(path).forall(isClustered)) + assert(lastOptimizeClusterBy(path) === """["id","d"]""") + checkData(path) + } + } + + for (columnMappingMode <- Seq("none", "name")) + test(s"files are grouped by the narrowest clustering column - " + + s"columnMapping: $columnMappingMode") { + withTempDir { dir => + val path = dir.getCanonicalPath + // Every file covers all values of `id` and `s`, but a single day of `d`. + createTable(path, clusterBy = "id, s, d", columnMappingMode) + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true") { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + assert(files(path).size === 4) + assert(files(path).forall(!isClustered(_))) + assert(dayRanges(path) === groupedByDay) + assert(lastOptimizeClusterBy(path) === "[]") + checkData(path) + } + } + + test("files are grouped by size if no clustering column is narrow") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, s") + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true") { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + assert(files(path).forall(!isClustered(_))) + assert(dayRanges(path) === groupedBySize) + checkData(path) + } + } + + for ((column, expected) <- Seq( + "d" -> groupedByDay, + "D" -> groupedByDay, + "`n`.d" -> groupedByDay, + // Every file has the same min `id`, so files are ordered by their max `id`, i.e. by size. + "id" -> groupedBySize)) + test(s"files are grouped by the configured clustering column - column: $column") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d, n.d") + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true", columnKey -> column) { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + assert(dayRanges(path) === expected) + checkData(path) + } + } + + test("configured column must be a clustering column") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true", columnKey -> "s") { + val e = intercept[DeltaAnalysisException] { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + checkError( + e, + "DELTA_INVALID_LIGHTWEIGHT_CLUSTERING_COLUMN", + parameters = Map( + "column" -> "s", + "config" -> columnKey, + "clusteringColumns" -> "id, d")) + } + } + } + + test("clustered files are not rewritten") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path, days = 0 until 4) + optimize(path, maxFileSize = 1L << 30) + val clusteredFiles = files(path).map(_.path).toSet + assert(files(path).forall(isClustered)) + + appendOneFilePerDay(path, days = 4 until 8) + val newFiles = files(path).filterNot(isClustered) + val sizes = newFiles.map(_.size).sorted + withSQLConf(enabledKey -> "true") { + // Room for two new files per bin. + optimize(path, maxFileSize = sizes.takeRight(2).sum) + } + val filesAfter = files(path) + assert(clusteredFiles.subsetOf(filesAfter.map(_.path).toSet)) + val compactedFiles = filesAfter.filterNot(f => clusteredFiles.contains(f.path)) + assert(compactedFiles.forall(!isClustered(_))) + assert(dayRanges(path, Some(compactedFiles.map(_.path).toSet)) === Set((4, 5), (6, 7))) + checkData(path) + } + } + + test("OPTIMIZE clusters once unclustered data reaches the threshold") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path) + val unclusteredBytes = files(path).map(_.size).sum + withSQLConf(enabledKey -> "true", thresholdKey -> unclusteredBytes.toString) { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + assert(files(path).forall(isClustered)) + assert(lastOptimizeClusterBy(path) === """["id","d"]""") + checkData(path) + } + } + + test("OPTIMIZE FULL always clusters") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true") { + optimize(path, maxFileSizeForTwoFileBins(path), full = true) + } + assert(files(path).forall(isClustered)) + checkData(path) + } + } + + test("lightweight compaction output is clustered later") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + appendOneFilePerDay(path) + withSQLConf(enabledKey -> "true") { + optimize(path, maxFileSizeForTwoFileBins(path)) + } + assert(files(path).forall(!isClustered(_))) + optimize(path, maxFileSize = 1L << 30) + assert(files(path).forall(isClustered)) + checkData(path) + } + } + + test("auto compaction runs lightweight compaction") { + withTempDir { dir => + val path = dir.getCanonicalPath + createTable(path, clusterBy = "id, d") + withSQLConf( + enabledKey -> "true", + DeltaSQLConf.DELTA_AUTO_COMPACT_ENABLED.key -> "true", + DeltaSQLConf.DELTA_AUTO_COMPACT_MIN_NUM_FILES.key -> rowsPerDay.size.toString) { + appendOneFilePerDay(path) + } + val lastCommit = DeltaLog.forTable(spark, path).history.getHistory(Some(1)).head + assert(lastCommit.operation === DeltaOperations.OPTIMIZE_OPERATION_NAME) + assert(lastCommit.operationParameters("auto") === "true") + assert(lastOptimizeClusterBy(path) === "[]") + assert(files(path).size < rowsPerDay.size) + assert(files(path).forall(!isClustered(_))) + checkData(path) + } + } + + test("relativeFileRange") { + import ClusteringColumnFileOrdering.relativeFileRange + def ranges(rs: (Any, Any)*): Seq[ValueRange] = + rs.map { case (min, max) => ValueRange(min, max) } + assert(relativeFileRange(ranges((1, 1), (2, 2), (3, 3))) === Some(0.0)) + assert(relativeFileRange(ranges((1, 2), (3, 4))) === Some(1.0 / 3)) + assert(relativeFileRange(ranges((1, 9), (1, 9))) === Some(1.0)) + assert(relativeFileRange(ranges(("a", "z"), ("b", "c"))) === Some((3.0 + 1.0) / 2 / 3)) + // Not enough files with statistics, or a single value. + assert(relativeFileRange(ranges((1, 2), (null, null))) === None) + assert(relativeFileRange(ranges((1, 1), (1, 1))) === None) + } +}