Manifest generation blocking when DVs are present

GitOrigin-RevId: 95d4ea5d36e55315ae07795021a0473850bd4f56

Author: vkorukanti

core/src/main/resources/error/delta-error-classes.json

@@ -1876,6 +1876,13 @@
     ],
     "sqlState" : "0A000"
   },
+  "DELTA_UNSUPPORTED_GENERATE_WITH_DELETION_VECTORS" : {
+    "message" : [
+      "The 'GENERATE symlink_format_manifest' command is not supported on table versions with deletion vectors.",
+      "If you need to generate manifests, consider disabling deletion vectors on this table using 'ALTER TABLE table SET TBLPROPERTIES (delta.enableDeletionVectors = false)'."
+    ],
+    "sqlState" : "0A000"
+  },
   "DELTA_UNSUPPORTED_INVARIANT_NON_STRUCT" : {
     "message" : [
       "Invariants on nested fields other than StructTypes are not supported."
@@ -2009,6 +2016,30 @@
     ],
     "sqlState" : "23001"
   },
+  "DELTA_VIOLATE_TABLE_PROPERTY_VALIDATION_FAILED" : {
+    "message" : [
+      "The validation of the properties of table <table> has been violated:"
+    ],
+    "subClass" : {
+      "EXISTING_DELETION_VECTORS_WITH_INCREMENTAL_MANIFEST_GENERATION" : {
+        "message" : [
+          "Symlink manifest generation is unsupported while deletion vectors are present in the table.",
+          "In order to produce a version of the table without deletion vectors, run 'REORG TABLE <table> APPLY (PURGE)'."
+        ]
+      },
+      "PERSISTENT_DELETION_VECTORS_IN_NON_PARQUET_TABLE" : {
+        "message" : [
+          "Persistent deletion vectors are only supported on Parquet-based Delta tables."
+        ]
+      },
+      "PERSISTENT_DELETION_VECTORS_WITH_INCREMENTAL_MANIFEST_GENERATION" : {
+        "message" : [
+          "Persistent deletion vectors and incremental symlink manifest generation are mutually exclusive."
+        ]
+      }
+    },
+    "sqlState" : "0A000"
+  },
   "DELTA_ZORDERING_COLUMN_DOES_NOT_EXIST" : {
     "message" : [
       "Z-Ordering column <columnName> does not exist in data schema."

core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -2644,6 +2644,10 @@ trait DeltaErrorsBase
       messageParameters = Array.empty,
       pos = 0)
   }
+
+  def generateNotSupportedWithDeletionVectors(): Throwable =
+    new DeltaCommandUnsupportedWithDeletionVectorsException(
+      errorClass = "DELTA_UNSUPPORTED_GENERATE_WITH_DELETION_VECTORS")
 }
 
 object DeltaErrors extends DeltaErrorsBase
@@ -2923,6 +2927,49 @@ class DeltaNoSuchTableException(
   override def getErrorClass: String = errorClass
 }
 
+class DeltaCommandUnsupportedWithDeletionVectorsException(
+  errorClass: String,
+  messageParameters: Array[String] = Array.empty)
+  extends UnsupportedOperationException(
+    DeltaThrowableHelper.getMessage(errorClass, messageParameters))
+    with DeltaThrowable {
+  override def getErrorClass: String = errorClass
+}
+
+sealed trait DeltaTablePropertyValidationFailedSubClass {
+  def tag: String
+  /** Can be overridden in case subclasses need the table name as well. */
+  def messageParameters(table: String): Array[String] = Array(table)
+}
+
+final object DeltaTablePropertyValidationFailedSubClass {
+  final case object PersistentDeletionVectorsWithIncrementalManifestGeneration
+    extends DeltaTablePropertyValidationFailedSubClass {
+    override val tag = "PERSISTENT_DELETION_VECTORS_WITH_INCREMENTAL_MANIFEST_GENERATION"
+  }
+  final case object ExistingDeletionVectorsWithIncrementalManifestGeneration
+    extends DeltaTablePropertyValidationFailedSubClass {
+    override val tag = "EXISTING_DELETION_VECTORS_WITH_INCREMENTAL_MANIFEST_GENERATION"
+    /** This subclass needs the table parameters in two places. */
+    override def messageParameters(table: String): Array[String] = Array(table, table)
+  }
+  final case object PersistentDeletionVectorsInNonParquetTable
+    extends DeltaTablePropertyValidationFailedSubClass {
+    override val tag = "PERSISTENT_DELETION_VECTORS_IN_NON_PARQUET_TABLE"
+  }
+}
+
+class DeltaTablePropertyValidationFailedException(
+    table: String,
+    subClass: DeltaTablePropertyValidationFailedSubClass)
+  extends RuntimeException(DeltaThrowableHelper.getMessage(
+    errorClass = "DELTA_VIOLATE_TABLE_PROPERTY_VALIDATION_FAILED",
+    messageParameters = subClass.messageParameters(table)))
+    with DeltaThrowable {
+  override def getErrorClass: String =
+    "DELTA_VIOLATE_TABLE_PROPERTY_VALIDATION_FAILED." + subClass.tag
+}
+
 /** Errors thrown around column mapping. */
 class ColumnMappingUnsupportedException(msg: String)
   extends UnsupportedOperationException(msg)

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

@@ -554,6 +554,9 @@ trait OptimisticTransactionImpl extends TransactionalWrite
       }
     }
 
+    if (spark.conf.get(DeltaSQLConf.DELTA_TABLE_PROPERTY_CONSTRAINTS_CHECK_ENABLED)) {
+      Protocol.assertTablePropertyConstraintsSatisfied(spark, metadata, snapshot)
+    }
 
     val needsProtocolUpdate = Protocol.checkProtocolRequirements(spark, metadata, protocol)
     if (needsProtocolUpdate.isDefined) {

core/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala

@@ -26,7 +26,7 @@ import scala.collection.mutable
 import scala.util.control.NonFatal
 
 import org.apache.spark.sql.delta._
-import org.apache.spark.sql.delta.constraints.Constraints
+import org.apache.spark.sql.delta.commands.DeletionVectorUtils
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.util.JsonUtils
 import com.fasterxml.jackson.annotation._
@@ -36,11 +36,10 @@ import com.fasterxml.jackson.databind._
 import com.fasterxml.jackson.databind.annotation.{JsonDeserialize, JsonSerialize}
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.sql.{Column, DataFrame, Dataset, Encoder, SparkSession}
+import org.apache.spark.sql.{Column, Encoder, SparkSession}
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.Literal
-import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataType, StructField, StructType}
 import org.apache.spark.util.Utils
 
@@ -339,6 +338,46 @@ object Protocol {
     }
   }
 
+  /**
+   * Verify that the table properties satisfy legality constraints. Throw an exception if not.
+   */
+  def assertTablePropertyConstraintsSatisfied(
+      spark: SparkSession,
+      metadata: Metadata,
+      snapshot: Snapshot): Unit = {
+    import DeltaTablePropertyValidationFailedSubClass._
+
+    val tableName = if (metadata.name != null) metadata.name else metadata.id
+
+    val configs = metadata.configuration.map { case (k, v) => k.toLowerCase(Locale.ROOT) -> v }
+    val dvsEnabled = {
+      val lowerCaseKey = DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key.toLowerCase(Locale.ROOT)
+      configs.get(lowerCaseKey).exists(_.toBoolean)
+    }
+    if (dvsEnabled && metadata.format.provider != "parquet") {
+      // DVs only work with parquet-based delta tables.
+      throw new DeltaTablePropertyValidationFailedException(
+        table = tableName,
+        subClass = PersistentDeletionVectorsInNonParquetTable)
+    }
+    val manifestGenerationEnabled = {
+      val lowerCaseKey = DeltaConfigs.SYMLINK_FORMAT_MANIFEST_ENABLED.key.toLowerCase(Locale.ROOT)
+      configs.get(lowerCaseKey).exists(_.toBoolean)
+    }
+    if (dvsEnabled && manifestGenerationEnabled) {
+      throw new DeltaTablePropertyValidationFailedException(
+        table = tableName,
+        subClass = PersistentDeletionVectorsWithIncrementalManifestGeneration)
+    }
+    if (manifestGenerationEnabled) {
+      // Only allow enabling this, if there are no DVs present.
+      if (!DeletionVectorUtils.isTableDVFree(spark, snapshot)) {
+        throw new DeltaTablePropertyValidationFailedException(
+          table = tableName,
+          subClass = ExistingDeletionVectorsWithIncrementalManifestGeneration)
+      }
+    }
+  }
 }
 
 /**
@@ -406,7 +445,8 @@ case class AddFile(
     // scalastyle:off
     val removedFile = RemoveFile(
       path, Some(timestamp), dataChange,
-      extendedFileMetadata = Some(true), partitionValues, Some(size), newTags
+      extendedFileMetadata = Some(true), partitionValues, Some(size), newTags,
+      deletionVector = deletionVector
     )
     // scalastyle:on
     removedFile

core/src/main/scala/org/apache/spark/sql/delta/commands/OptimizeTableCommand.scala

@@ -306,10 +306,13 @@ class OptimizeExecutor(
           bins += currentBin.toVector
         }
 
+        bins.filter { bin =>
+          bin.size > 1 || // bin has more than one file or
+          (bin.size == 1 && bin(0).deletionVector != null) || // single file in the bin has a DV or
+          isMultiDimClustering // multi-clustering
+        }
+
         bins.map(b => (partition, b))
-          // select bins that have at least two files or in case of multi-dim clustering
-          // select all bins
-          .filter(_._2.size > 1 || isMultiDimClustering)
     }
   }
 

core/src/main/scala/org/apache/spark/sql/delta/hooks/GenerateSymlinkManifest.scala

@@ -21,6 +21,7 @@ import java.net.URI
 
 import org.apache.spark.sql.delta._
 import org.apache.spark.sql.delta.actions._
+import org.apache.spark.sql.delta.commands.DeletionVectorUtils.isTableDVFree
 import org.apache.spark.sql.delta.metering.DeltaLogging
 import org.apache.spark.sql.delta.storage.LogStore
 import org.apache.spark.sql.delta.util.DeltaFileOperations
@@ -78,6 +79,7 @@ trait GenerateSymlinkManifestImpl extends PostCommitHook with DeltaLogging with
   override def handleError(error: Throwable, version: Long): Unit = {
     error match {
       case e: ColumnMappingUnsupportedException => throw e
+      case e: DeltaCommandUnsupportedWithDeletionVectorsException => throw e
       case _ =>
         throw DeltaErrors.postCommitHookFailedException(this, version, name, error)
     }
@@ -178,6 +180,7 @@ trait GenerateSymlinkManifestImpl extends PostCommitHook with DeltaLogging with
       spark: SparkSession,
       deltaLog: DeltaLog): Unit = {
     val snapshot = deltaLog.update(stalenessAcceptable = false)
+    assertTableIsDVFree(spark, snapshot)
     generateFullManifestWithSnapshot(spark, deltaLog, snapshot)
   }
 
@@ -241,6 +244,12 @@ trait GenerateSymlinkManifestImpl extends PostCommitHook with DeltaLogging with
     recordDeltaEvent(deltaLog, s"$FULL_MANIFEST_OP_TYPE.stats", data = stats)
   }
 
+  protected def assertTableIsDVFree(spark: SparkSession, snapshot: Snapshot): Unit = {
+    if (!isTableDVFree(spark, snapshot)) {
+      throw DeltaErrors.generateNotSupportedWithDeletionVectors()
+    }
+  }
+
   /**
    * Write the manifest files and return the partition relative paths of the manifests written.
    *

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

@@ -1091,6 +1091,15 @@ trait DeltaSQLConfBase {
       .checkValue(_ >= 0, "maxDeletedRowsRatio must be in range [0.0, 1.0]")
       .checkValue(_ <= 1, "maxDeletedRowsRatio must be in range [0.0, 1.0]")
       .createWithDefault(0.05d)
+
+  val DELTA_TABLE_PROPERTY_CONSTRAINTS_CHECK_ENABLED =
+    buildConf("tablePropertyConstraintsCheck.enabled")
+      .internal()
+      .doc(
+        """Check that all table-properties satisfy validity constraints.
+          |Only change this for testing!""".stripMargin)
+      .booleanConf
+      .createWithDefault(true)
 }
 
 object DeltaSQLConf extends DeltaSQLConfBase

core/src/test/scala/org/apache/spark/sql/delta/DeletionVectorsTestUtils.scala

@@ -104,6 +104,13 @@ trait DeletionVectorsTestUtils extends QueryTest with SharedSparkSession {
     }
   }
 
+  /** Enable persistent Deletion Vectors in a Delta table. */
+  def enableDeletionVectorsInTable(tablePath: Path, enable: Boolean): Unit =
+    spark.sql(
+      s"""ALTER TABLE delta.`$tablePath`
+         |SET TBLPROPERTIES ('${DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key}' = '$enable')
+         |""".stripMargin)
+
   // ======== HELPER METHODS TO WRITE DVs ==========
   /** Helper method to remove the specified rows in the given file using DVs */
   protected def removeRowsFromFileUsingDV(