[jira] [Updated] (SPARK-59631) Persisted V1 time-travel relations are recached after table writes

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-59631?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated SPARK-59631:
-----------------------------------
    Labels: pull-request-available  (was: )

> Persisted V1 time-travel relations are recached after table writes
> ------------------------------------------------------------------
>
>                 Key: SPARK-59631
>                 URL: https://issues.apache.org/jira/browse/SPARK-59631
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 4.1.0, 4.2.0
>            Reporter: Joel Robin
>            Priority: Major
>              Labels: pull-request-available
>
> h2. Description
> A persisted DataFrame reading a fixed table version through the V1 data source path is unnecessarily recached after a later write advances the live table.
> The pinned snapshot is immutable, but recaching re-executes its entire plan. This can repeat expensive or non-idempotent UDFs even though the DataFrame remains registered as cached.
> h2. Reproduction
> Tested with unpatched Apache Spark 4.2.0, Scala 2.13, Java 17, and Delta Lake master commit {{052429d500f64f7b3d482f28092b2eb066d20cdb}}. Delta V2 mode is disabled to exercise the affected V1 relation path.
> Compile and run the following standalone application with the local Spark and Delta classpath:
> {code:scala}
> import java.nio.file.Files
> import java.util.UUID
> import org.apache.spark.sql.SparkSession
> import org.apache.spark.sql.functions.{col, udf}
> import org.apache.spark.storage.StorageLevel
> object DeltaTimeTravelCacheRepro {
>   def main(args: Array[String]): Unit = {
>     val root = Files.createTempDirectory("delta-time-travel-cache-repro-")
>     val table = "tt_cache_" + UUID.randomUUID().toString.replace("-", "")
>     val spark = SparkSession.builder()
>       .appName("DeltaTimeTravelCacheRepro")
>       .master("local[2]")
>       .config("spark.ui.enabled", "false")
>       .config("spark.sql.shuffle.partitions", "2")
>       .config("spark.sql.warehouse.dir", root.resolve("warehouse").toUri.toString)
>       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
>       .config(
>         "spark.sql.catalog.spark_catalog",
>         "org.apache.spark.sql.delta.catalog.DeltaCatalog")
>       .config("spark.databricks.delta.v2.enableMode", "NONE")
>       .getOrCreate()
>     var cached: Option[org.apache.spark.sql.DataFrame] = None
>     try {
>       spark.range(10).write.format("delta").saveAsTable(table)
>       val udfCalls = spark.sparkContext.longAccumulator("udf-calls")
>       val expensiveUdf = udf { id: Long =>
>         udfCalls.add(1L)
>         s"$id-${UUID.randomUUID()}"
>       }.asNondeterministic()
>       val pinned = spark.read
>         .option("versionAsOf", 0)
>         .table(table)
>         .withColumn("token", expensiveUdf(col("id")))
>         .persist()
>       cached = Some(pinned)
>       // Materialize the cache before advancing the live table.
>       val before = pinned.orderBy("id").select("token").collect().toSeq
>       val callsBefore = udfCalls.value
>       // This creates version 1. The pinned DataFrame still reads version 0.
>       spark.range(10, 20)
>         .write
>         .format("delta")
>         .mode("append")
>         .saveAsTable(table)
>       val after = pinned.orderBy("id").select("token").collect().toSeq
>       val callsAfter = udfCalls.value
>       val pinnedCount = pinned.count()
>       val liveCount = spark.table(table).count()
>       println(s"is_cached=${pinned.storageLevel != StorageLevel.NONE}")
>       println(s"udf_calls=$callsBefore->$callsAfter")
>       println(s"tokens_same=${before == after}")
>       println(s"pinned_count=$pinnedCount")
>       println(s"live_count=$liveCount")
>       assert(callsBefore == 10)
>       assert(callsAfter == callsBefore)
>       assert(after == before)
>       assert(pinnedCount == 10)
>       assert(liveCount == 20)
>     } finally {
>       cached.foreach(_.unpersist(blocking = true))
>       spark.sql(s"DROP TABLE IF EXISTS $table")
>       spark.stop()
>     }
>   }
> }
> {code}
> The nondeterministic Scala UDF is a stand-in for the reported pandas UDF that performs LLM calls. It isolates the cache invalidation behavior without requiring Python or Arrow.
> h2. Actual Result
> On unpatched Spark 4.2.0:
> {noformat}
> is_cached=true
> udf_calls=10->20
> tokens_same=false
> pinned_count=10
> live_count=20
> {noformat}
> The second action recomputes the cached plan and invokes the UDF again. The cache entry remains registered, which explains why the DataFrame still reports as cached while its materialized buffers have been discarded.
> h2. Expected Result
> {noformat}
> is_cached=true
> udf_calls=10->10
> tokens_same=true
> pinned_count=10
> live_count=20
> {noformat}
> A write that advances the live table should refresh live-table cache entries but should not invalidate a cache entry for an immutable, version-pinned snapshot. Explicit operations such as {{REFRESH TABLE}}, {{refreshByPath}}, or uncache should retain their existing ability to invalidate pinned snapshots.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org