Joel Robin created SPARK-59631:
----------------------------------
Summary: 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.2.0, 4.1.0
Reporter: Joel Robin
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: [email protected]
For additional commands, e-mail: [email protected]
