[SPARK-54341][SQL] Remember TimeTravelSpec for tables loaded via TableProvider

### What changes were proposed in this pull request?

This PR adds logic to remember `TimeTravelSpec` for tables loaded via `TableProvider`.

### Why are the changes needed?

These changes are needed to align `TableProvider` with other places that remember time travel specs.

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

No.

### How was this patch tested?

This PR comes with tests.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#53048 from aokolnychyi/spark-54341.

Authored-by: Anton Okolnychyi <aokolnychyi@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: aokolnychyi

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala

@@ -115,7 +115,7 @@ private[sql] object DataSourceV2Utils extends Logging {
     val finalOptions = sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
       optionsWithPath.originalMap
     val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
-    val (table, catalog, ident) = provider match {
+    val (table, catalog, ident, timeTravelSpec) = provider match {
       case _: SupportsCatalogOptions if userSpecifiedSchema.nonEmpty =>
         throw new IllegalArgumentException(
           s"$source does not support user specified schema. Please don't specify the schema.")
@@ -141,16 +141,17 @@ private[sql] object DataSourceV2Utils extends Logging {
         }
         val timeTravel = TimeTravelSpec.create(
           timeTravelTimestamp, timeTravelVersion, conf.sessionLocalTimeZone)
-        (CatalogV2Util.getTable(catalog, ident, timeTravel), Some(catalog), Some(ident))
+        val tbl = CatalogV2Util.getTable(catalog, ident, timeTravel)
+        (tbl, Some(catalog), Some(ident), timeTravel)
       case _ =>
         // TODO: Non-catalog paths for DSV2 are currently not well defined.
         val tbl = DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
-        (tbl, None, None)
+        (tbl, None, None, None)
     }
     import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
     table match {
       case _: SupportsRead if table.supports(BATCH_READ) =>
-        Option(DataSourceV2Relation.create(table, catalog, ident, dsOptions))
+        Option(DataSourceV2Relation.create(table, catalog, ident, dsOptions, timeTravelSpec))
       case _ => None
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/connector/SupportsCatalogOptionsSuite.scala

@@ -26,8 +26,8 @@ import scala.util.Try
 import org.scalatest.BeforeAndAfter
 
 import org.apache.spark.SparkException
-import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, SaveMode}
-import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, TableAlreadyExistsException}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, QueryTest, SaveMode}
+import org.apache.spark.sql.catalyst.analysis.{AsOfTimestamp, AsOfVersion, NoSuchTableException, TableAlreadyExistsException, TimeTravelSpec}
 import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, OverwriteByExpression}
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
 import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, InMemoryTableCatalog, SupportsCatalogOptions, TableCatalog}
@@ -305,8 +305,13 @@ class SupportsCatalogOptionsSuite extends QueryTest with SharedSparkSession with
         .mode(SaveMode.Overwrite).save()
 
       // load with version
-      checkAnswer(load("t", Some(catalogName), version = Some("Snapshot123456789")), df1.toDF())
-      checkAnswer(load("t", Some(catalogName), version = Some("2345678910")), df2.toDF())
+      val readDF1 = load("t", Some(catalogName), version = Some("Snapshot123456789"))
+      checkAnswer(readDF1, df1.toDF())
+      checkTimeTravel(readDF1, expectedTimeTravelSpec = AsOfVersion("Snapshot123456789"))
+
+      val readDF2 = load("t", Some(catalogName), version = Some("2345678910"))
+      checkAnswer(readDF2, df2.toDF())
+      checkTimeTravel(readDF2, expectedTimeTravelSpec = AsOfVersion("2345678910"))
     }
 
     val ts1 = DateTimeUtils.stringToTimestampAnsi(
@@ -330,16 +335,26 @@ class SupportsCatalogOptionsSuite extends QueryTest with SharedSparkSession with
         .mode(SaveMode.Overwrite).save()
 
       // load with timestamp
-      checkAnswer(load("t", Some(catalogName), version = None,
-        timestamp = Some("2019-01-29 00:37:58")), df3.toDF())
-      checkAnswer(load("t", Some(catalogName), version = None,
-        timestamp = Some("2021-01-29 00:37:58")), df4.toDF())
+      val readDF3 = load("t", Some(catalogName), version = None,
+        timestamp = Some("2019-01-29 00:37:58"))
+      checkAnswer(readDF3, df3.toDF())
+      checkTimeTravel(readDF3, expectedTimeTravelSpec = AsOfTimestamp(ts1))
+
+      val readDF4 = load("t", Some(catalogName), version = None,
+        timestamp = Some("2021-01-29 00:37:58"))
+      checkAnswer(readDF4, df4.toDF())
+      checkTimeTravel(readDF4, expectedTimeTravelSpec = AsOfTimestamp(ts2))
 
       // load with timestamp in number format
-      checkAnswer(load("t", Some(catalogName), version = None,
-        timestamp = Some(MICROSECONDS.toSeconds(ts1).toString)), df3.toDF())
-      checkAnswer(load("t", Some(catalogName), version = None,
-        timestamp = Some(MICROSECONDS.toSeconds(ts2).toString)), df4.toDF())
+      val readDF5 = load("t", Some(catalogName), version = None,
+        timestamp = Some(MICROSECONDS.toSeconds(ts1).toString))
+      checkAnswer(readDF5, df3.toDF())
+      checkTimeTravel(readDF5, expectedTimeTravelSpec = AsOfTimestamp(ts1))
+
+      val readDF6 = load("t", Some(catalogName), version = None,
+        timestamp = Some(MICROSECONDS.toSeconds(ts2).toString))
+      checkAnswer(readDF6, df4.toDF())
+      checkTimeTravel(readDF6, expectedTimeTravelSpec = AsOfTimestamp(ts2))
     }
 
     val e = intercept[AnalysisException] {
@@ -360,6 +375,11 @@ class SupportsCatalogOptionsSuite extends QueryTest with SharedSparkSession with
     assert(v2.catalog.exists(_ == catalogPlugin))
   }
 
+  private def checkTimeTravel(ds: Dataset[_], expectedTimeTravelSpec: TimeTravelSpec): Unit = {
+    val relation = ds.logicalPlan.asInstanceOf[DataSourceV2Relation]
+    assert(relation.timeTravelSpec.contains(expectedTimeTravelSpec))
+  }
+
   private def load(
       name: String,
       catalogOpt: Option[String],