v1

delta-io · Feb 28, 2025 · 0258090 · 0258090
1 parent 9e17c07
commit 0258090
Show file tree

Hide file tree

Showing 4 changed files with 367 additions and 5 deletions.
diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/logging/DeltaLogKeys.scala b/spark/src/main/scala/org/apache/spark/sql/delta/logging/DeltaLogKeys.scala
@@ -64,6 +64,7 @@ trait DeltaLogKeysBase {
   case object END_OFFSET extends LogKeyShims
   case object END_VERSION extends LogKeyShims
   case object ERROR extends LogKeyShims
+  case object EXCEPTION extends LogKeyShims
   case object EXECUTOR_ID extends LogKeyShims
   case object EXPR extends LogKeyShims
   case object FILE_INDEX extends LogKeyShims

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
@@ -480,6 +480,15 @@ trait DeltaSQLConfBase {
       .checkValue(_ >= 0, "DVTombstoneCountThreshold must not be negative.")
       .createWithDefault(10000)
 
+  val FAST_DROP_FEATURE_STREAMING_ALWAYS_VALIDATE_PROTOCOL =
+    buildConf("tableFeatures.dev.fastDropFeature.alwaysValidateProtocolInStreaming.enabled")
+      .internal()
+      .doc(
+        """Whether to validate the protocol when starting a stream from arbitrary
+          |versions.""".stripMargin)
+      .booleanConf
+      .createWithDefault(true)
+
   val DELTA_MAX_SNAPSHOT_LINEAGE_LENGTH =
     buildConf("maxSnapshotLineageLength")
       .internal()

diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala
@@ -21,6 +21,7 @@ import java.io.FileNotFoundException
 import java.sql.Timestamp
 
 import scala.util.{Failure, Success, Try}
+import scala.util.control.NonFatal
 import scala.util.matching.Regex
 
 import org.apache.spark.sql.delta._
@@ -1360,9 +1361,15 @@ case class DeltaSource(
         case StartingVersionLatest =>
           deltaLog.update().version + 1
         case StartingVersion(version) =>
-          // when starting from a given version, we don't need the snapshot of this version. So
-          // `mustBeRecreatable` is set to `false`.
-          deltaLog.history.checkVersionExists(version, mustBeRecreatable = false, allowOutOfRange)
+          if (!DeltaSource.validateProtocolAt(spark, deltaLog, version)) {
+            // When starting from a given version, we don't require that the snapshot of this
+            // version can be reconstructed, even though the input table is technically in an
+            // inconsistent state. If the snapshot cannot be reconstructed, then the protocol
+            // check is skipped, so this is technically not safe, but we keep it this way for
+            // historical reasons.
+            deltaLog.history.checkVersionExists(
+              version, mustBeRecreatable = false, allowOutOfRange)
+          }
           version
       }
       Some(v)
@@ -1384,7 +1391,33 @@ case class DeltaSource(
 
 }
 
-object DeltaSource {
+object DeltaSource extends DeltaLogging {
+  /**
+   * Validate the protocol at a given version. If the snapshot reconstruction fails for any other
+   * reason than table feature exception, we suppress it. This allows to fallback to previous
+   * behavior where the starting version/timestamp was not mandatory to point to reconstructable
+   * snapshot.
+   *
+   * Returns true when the validation was performed and succeeded.
+   */
+  def validateProtocolAt(spark: SparkSession, deltaLog: DeltaLog, version: Long): Boolean = {
+    val alwaysValidateProtocol = spark.sessionState.conf.getConf(
+      DeltaSQLConf.FAST_DROP_FEATURE_STREAMING_ALWAYS_VALIDATE_PROTOCOL)
+    if (!alwaysValidateProtocol) return false
+
+    try {
+      // We attempt to construct a snapshot at the startingVersion in order to validate the
+      // protocol. If snapshot reconstruction fails, fall back to the old behavior where the
+      // only requirement was for the commit to exist.
+      deltaLog.getSnapshotAt(version)
+      return true
+    } catch {
+      case e: DeltaUnsupportedTableFeatureException => throw e
+      case NonFatal(e) => // Suppress rest errors.
+        logWarning(log"Protocol validation failed with '${MDC(DeltaLogKeys.EXCEPTION, e)}'.")
+    }
+    false
+  }
 
   /**
    * - If a commit version exactly matches the provided timestamp, we return it.
@@ -1414,6 +1447,7 @@ object DeltaSource {
       mustBeRecreatable = false,
       canReturnEarliestCommit = true)
     if (commit.timestamp >= timestamp.getTime) {
+      validateProtocolAt(spark, deltaLog, commit.version)
       // Find the commit at the `timestamp` or the earliest commit
       commit.version
     } else {
@@ -1423,7 +1457,9 @@ object DeltaSource {
       //
       // Note2: In the use case of [[CDCReader]] timestamp passed in can exceed the latest commit
       // timestamp, caller doesn't expect exception, and can handle the non-existent version.
-      if (commit.version + 1 <= deltaLog.unsafeVolatileSnapshot.version || canExceedLatest) {
+      val latestNotExceeded = commit.version + 1 <= deltaLog.unsafeVolatileSnapshot.version
+      if (latestNotExceeded || canExceedLatest) {
+        if (latestNotExceeded) validateProtocolAt(spark, deltaLog, commit.version + 1)
         commit.version + 1
       } else {
         val commitTs = new Timestamp(commit.timestamp)