Refresh presigned urls for delta sharing streaming, non cdf queries

spark/src/main/scala/io/delta/sharing/spark/DeltaSharingSource.scala

@@ -17,8 +17,11 @@
 package io.delta.sharing.spark
 
 // scalastyle:off import.ordering.noEmptyLine
+import java.lang.ref.WeakReference
+
 import scala.collection.mutable.ArrayBuffer
 
+import org.apache.spark.delta.sharing.CachedTableManager
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{DataFrame, DeltaSharingScanUtils, SparkSession}
 import org.apache.spark.sql.connector.read.streaming
@@ -235,6 +238,13 @@ case class DeltaSharingSource(
       // If isStartingVersion is true, it means to fetch the snapshot at the fromVersion, which may
       // include table changes from previous versions.
       val tableFiles = deltaLog.client.getFiles(deltaLog.table, Nil, None, Some(fromVersion), None)
+      latestRefreshFunc = () => {
+        deltaLog.client.getFiles(
+          deltaLog.table, Nil, None, Some(fromVersion), None
+        ).files.map { f =>
+          f.id -> f.url
+        }.toMap
+      }
 
       val numFiles = tableFiles.files.size
       tableFiles.files.sortWith(fileActionCompareFunc).zipWithIndex.foreach {
@@ -260,6 +270,11 @@ case class DeltaSharingSource(
       // If isStartingVersion is false, it means to fetch table changes since fromVersion, not
       // including files from previous versions.
       val tableFiles = deltaLog.client.getFiles(deltaLog.table, fromVersion)
+      latestRefreshFunc = () => {
+        deltaLog.client.getFiles(deltaLog.table, fromVersion).addFiles.map { a =>
+          a.id -> a.url
+        }.toMap
+      }
       val allAddFiles = verifyStreamHygieneAndFilterAddFiles(tableFiles).groupBy(a => a.version)
       for (v <- fromVersion to currentLatestVersion) {
 
@@ -479,9 +494,14 @@ case class DeltaSharingSource(
       val add = indexedFile.add
       AddFile(add.url, add.id, add.partitionValues, add.size, add.stats)
     }
+    val idToUrl = addFilesList.map { add =>
+      add.id -> add.url
+    }.toMap
 
     val params = new RemoteDeltaFileIndexParams(spark, initSnapshot)
     val fileIndex = new RemoteDeltaBatchFileIndex(params, addFilesList)
+    CachedTableManager.INSTANCE.register(
+      params.path.toString, idToUrl, Seq(new WeakReference(fileIndex)), latestRefreshFunc)
 
     val relation = HadoopFsRelation(
       fileIndex,

spark/src/main/scala/io/delta/sharing/spark/RemoteDeltaFileIndex.scala

@@ -179,21 +179,8 @@ private[sharing] case class RemoteDeltaBatchFileIndex(
   override def listFiles(
     partitionFilters: Seq[Expression],
     dataFilters: Seq[Expression]): Seq[PartitionDirectory] = {
-    // TODO(lin.zhou): Actually refresh the presigned url in the cache instead of just using
-    // getIdToUrlMap
-    CachedTableManager.INSTANCE
-      .register(params.path.toString, getIdToUrlMap, Seq(new WeakReference(this)), () => {
-        getIdToUrlMap
-      })
-
     // We ignore partition filters for list files, since the delta sharing server already
     // parforms the filters.
     makePartitionDirectories(addFiles)
   }
-
-  private[sharing] def getIdToUrlMap : Map[String, String] = {
-    addFiles.map { add =>
-      add.id -> add.url
-    }.toMap
-  }
 }