[Spark] Fixes options-based time travel with timestamps (delta-io#4706)

#### Which Delta project/connector is this regarding?

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

This PR fixes a bug that occurs when trying to time travel on a Delta
table using Spark's dataframe API and timestamps, e.g. through
spark.read.option("timestampAsOf",
"some-timestamp").table("some-table"). For some input timestamps, this
can lead to the time travel specification being defined twice (once
through the DeltaTableV2 and once through the options), which is not
allowed. To work around this problem, there is a check for allowing two
time travel specifications in case they are equal.

However, in cases where the timestamp in the options is specified with
microsecond precision or without any milliseconds, the equality check
fails because the time travel specification in the DeltaTableV2 adds
`.0` milliseconds by default but drops the microseconds and so the
timestamp in the time travel specification stored as part of
DeltaTableV2 is different from the one defined in the options. The fix
in this PR normalizes the comparison to ensure that timestamps from both
specifications are compared correctly.

## How was this patch tested?

Added unit tests.

## Does this PR introduce _any_ user-facing changes?

No

---------

Co-authored-by: Lukas Rupprecht <[email protected]>
Co-authored-by: Allison Portis <[email protected]>

Author: 3 people

spark/src/main/scala-spark-3.5/shims/DeltaTimeTravelSpecShims.scala

@@ -16,6 +16,8 @@
 
 package org.apache.spark.sql.delta
 
+import org.apache.spark.sql.SparkSession
+
 object DeltaTimeTravelSpecShims {
 
   /**
@@ -36,6 +38,7 @@ object DeltaTimeTravelSpecShims {
    * @param newSpecOpt: The new [[DeltaTimeTravelSpec]] to be applied to the table
    */
   def validateTimeTravelSpec(
+      spark: SparkSession,
       currSpecOpt: Option[DeltaTimeTravelSpec],
       newSpecOpt: Option[DeltaTimeTravelSpec]): Unit = {
     if (currSpecOpt.nonEmpty && newSpecOpt.nonEmpty) {

spark/src/main/scala-spark-master/shims/DeltaTimeTravelSpecShims.scala

@@ -16,6 +16,8 @@
 
 package org.apache.spark.sql.delta
 
+import org.apache.spark.sql.SparkSession
+
 object DeltaTimeTravelSpecShims {
 
   /**
@@ -36,10 +38,13 @@ object DeltaTimeTravelSpecShims {
    * @param newSpecOpt: The new [[DeltaTimeTravelSpec]] to be applied to the table
    */
   def validateTimeTravelSpec(
+      spark: SparkSession,
       currSpecOpt: Option[DeltaTimeTravelSpec],
       newSpecOpt: Option[DeltaTimeTravelSpec]): Unit = (currSpecOpt, newSpecOpt) match {
     case (Some(currSpec), Some(newSpec))
-      if currSpec.version != newSpec.version || currSpec.timestamp != newSpec.timestamp =>
+      if currSpec.version != newSpec.version  ||
+        currSpec.getTimestampOpt(spark.sessionState.conf).map(_.getTime) !=
+          newSpec.getTimestampOpt(spark.sessionState.conf).map(_.getTime) =>
         throw DeltaErrors.multipleTimeTravelSyntaxUsed
     case _ =>
   }

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

@@ -70,6 +70,14 @@ case class DeltaTimeTravelSpec(
     }
     DateTimeUtils.toJavaTimestamp(castResult.asInstanceOf[java.lang.Long])
   }
+
+  /**
+   * Compute the timestamp to use for time travelling the relation from the given expression for
+   * the given time zone if this spec has a timestamp defined.
+   */
+  def getTimestampOpt(conf: SQLConf): Option[Timestamp] = {
+    timestamp.map(_ => getTimestamp(conf))
+  }
 }
 
 object DeltaTimeTravelSpec {

spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala

@@ -332,6 +332,7 @@ class DeltaTableV2 private[delta](
 
     // Spark 4.0 and 3.5 handle time travel options differently.
     DeltaTimeTravelSpecShims.validateTimeTravelSpec(
+      spark,
       currSpecOpt = timeTravelOpt,
       newSpecOpt = ttSpec)
 

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

@@ -822,6 +822,32 @@ class DeltaTimeTravelSuite extends QueryTest
         Row(1) :: Row(1) :: Row(2) :: Nil)
     }
   }
+
+  test("Dataframe-based time travel works with different timestamp precisions") {
+    val tblName = "test_tab"
+    withTable(tblName) {
+      sql(s"CREATE TABLE spark_catalog.default.$tblName (a int) USING DELTA")
+      // Ensure that the current timestamp is different from the one in the table.
+      Thread.sleep(1000)
+      // Microsecond precision timestamp.
+      val current_time_micros = spark.sql("SELECT current_timestamp() as ts")
+        .select($"ts".cast("string"))
+        .head().getString(0)
+      // Millisecond precision timestamp.
+      val current_time_millis = new Timestamp(System.currentTimeMillis())
+      // Second precision timestamp.
+      val sdf = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
+      val current_time_seconds = sdf.format(new java.sql.Timestamp(System.currentTimeMillis()))
+
+      sql(s"INSERT INTO spark_catalog.default.$tblName VALUES (1)")
+      checkAnswer(spark.read.option("timestampAsOf", current_time_micros)
+        .table(s"spark_catalog.default.$tblName"), Seq.empty)
+      checkAnswer(spark.read.option("timestampAsOf", current_time_millis.toString)
+        .table(s"spark_catalog.default.$tblName"), Seq.empty)
+      checkAnswer(spark.read.option("timestampAsOf", current_time_seconds)
+        .table(s"spark_catalog.default.$tblName"), Seq.empty)
+    }
+  }
 }
 
 class DeltaTimeTravelWithCatalogOwnedBatch1Suite extends DeltaTimeTravelSuite {