You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "joelrobin18 (via GitHub)" <gi...@apache.org> on 2026/09/18 08:36:06 UTC
[PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
joelrobin18 opened a new pull request, #58907:
URL: https://github.com/apache/spark/pull/58907
### What changes were proposed in this pull request?
This PR adds `FileIndex.isTimeTravel`, defaulting to `false`, so V1 file data sources can
identify immutable time-travel snapshots.
`CacheManager` honors this signal during write-driven cache invalidation by table name and path.
`InsertIntoHadoopFsRelationCommand` marks its path-based recache as write-driven. Existing explicit
refresh operations continue to include time-travel snapshots.
### Why are the changes needed?
[SPARK-53732](https://issues.apache.org/jira/browse/SPARK-53732) changed write-driven cache
invalidation from plan matching to table-name matching with `includeTimeTravel = false`. V2
relations honor that flag through `DataSourceV2Relation.timeTravelSpec`, but V1
`LogicalRelationWithTable` was matched only by table name.
For providers that lower a pinned read to `LogicalRelation(HadoopFsRelation(FileIndex))`, the
time-travel state survives only in the `FileIndex`. A write to the live table therefore cleared and
rebuilt the pinned cache. The DataFrame remained registered as cached, but the next action
re-executed its plan, including expensive or non-idempotent UDFs.
The new signal closes this V1 gap without coupling Spark to a specific data source.
JIRA: [SPARK-59631](https://issues.apache.org/jira/browse/SPARK-59631)
### Does this PR introduce _any_ user-facing change?
Yes. For V1 data sources that identify an immutable snapshot, appending to or overwriting the live
table no longer recomputes an already materialized cache for that snapshot. Live-table caches are
still refreshed after writes, and explicit refresh or uncache operations still include pinned
snapshots.
Before:
```text
is_cached=true
udf_calls=10->20
tokens_same=false
```
After the coordinated Spark and provider changes:
```text
is_cached=true
udf_calls=10->10
tokens_same=true
```
### How was this patch tested?
Added `V1TimeTravelCacheSuite` with six cases covering name- and path-based invalidation for pinned
and live V1 relations, including automatic write invalidation and explicit refresh.
```bash
build/sbt "sql/testOnly org.apache.spark.sql.V1TimeTravelCacheSuite"
```
All six tests passed. The existing V2 time-travel cache test, SQL MiMa, and main/test Scalastyle
checks also passed.
End-to-end testing reproduced `10->20` UDF executions with unpatched Spark and preserved `10->10`
with the coordinated Spark/provider changes. The provider integration matrix passed all four tests
in both `AUTO` and `NONE` modes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (gpt-5.6-sol)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58907:
URL: https://github.com/apache/spark/pull/58907#discussion_r4053329671
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -582,8 +593,22 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* `HadoopFsRelation` node(s) as part of its logical plan.
*/
def recacheByPath(spark: SparkSession, resourcePath: Path, fs: FileSystem): Unit = {
+ recacheByPath(spark, resourcePath, fs, includeTimeTravel = true)
+ }
+
+ /**
+ * Tries to re-cache all the cache entries that contain `resourcePath` in one or more
Review Comment:
**Nit (P3):** This overload says it tries to recache all matching entries, but `includeTimeTravel = false` intentionally leaves matching immutable snapshots loaded through both the V2 relation guard and `FileIndex.isTimeTravel`. Please document the parameter's inclusive/exclusive meaning and qualify the all-entries claim.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala:
##########
@@ -209,7 +209,11 @@ case class InsertIntoHadoopFsRelationCommand(
// refresh cached files in FileIndex
fileIndex.foreach(_.refresh())
// refresh data cache if table is cached
- sparkSession.sharedState.cacheManager.recacheByPath(sparkSession, outputPath, fs)
+ sparkSession.sharedState.cacheManager.recacheByPath(
+ sparkSession,
+ outputPath,
+ fs,
+ includeTimeTravel = false)
Review Comment:
**Non-blocking (P2):** The added `V1TimeTravelCacheSuite` invokes `recacheByPath(..., includeTimeTravel = false)` directly; it never executes this production caller. Removing or changing this argument would restore inclusive invalidation after real V1 writes while all six new tests still pass. Please add a regression that performs a real V1 file write with an already materialized immutable-snapshot cache and verifies that the snapshot stays loaded while a live cache refreshes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -592,16 +617,25 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* in the plan. If found, we refresh the metadata and return true. Otherwise, this method returns
* false.
*/
- private def lookupAndRefresh(plan: LogicalPlan, fs: FileSystem, qualifiedPath: Path): Boolean = {
+ private def lookupAndRefresh(
+ plan: LogicalPlan,
+ fs: FileSystem,
+ qualifiedPath: Path,
+ includeTimeTravel: Boolean): Boolean = {
plan match {
case lr: LogicalRelation => lr.relation match {
case hr: HadoopFsRelation =>
- refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath)
+ refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath, includeTimeTravel)
case _ => false
}
- case ExtractV2Table(fileTable: FileTable) =>
- refreshFileIndexIfNecessary(fileTable.fileIndex, fs, qualifiedPath)
+ case relation @ ExtractV2Table(fileTable: FileTable)
+ if includeTimeTravel || relation.timeTravelSpec.isEmpty =>
Review Comment:
**Non-blocking (P2):** This is a new V2 path-invalidation exclusion, but the added suite only builds V1 `LogicalRelation` fixtures and the existing V2 time-travel test exercises `recacheTableOrView`. Please add path-based coverage that reaches an `ExtractV2Table` time-travel relation, distinguishes write-driven from inclusive refresh, and fails if this `timeTravelSpec` guard is removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #58907: [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes
URL: https://github.com/apache/spark/pull/58907
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "joelrobin18 (via GitHub)" <gi...@apache.org>.
joelrobin18 commented on PR #58907:
URL: https://github.com/apache/spark/pull/58907#issuecomment-5771754796
@cloud-fan Updated the test suite. Could you please re-review? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "joelrobin18 (via GitHub)" <gi...@apache.org>.
joelrobin18 commented on code in PR #58907:
URL: https://github.com/apache/spark/pull/58907#discussion_r4068737692
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -582,8 +593,24 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* `HadoopFsRelation` node(s) as part of its logical plan.
*/
def recacheByPath(spark: SparkSession, resourcePath: Path, fs: FileSystem): Unit = {
+ recacheByPath(spark, resourcePath, fs, includeTimeTravel = true)
+ }
+
+ /**
+ * Tries to re-cache matching entries whose logical plans contain a file-backed relation under
+ * `resourcePath`. If `includeTimeTravel` is false, immutable time-travel relations are ignored
+ * when deciding whether an entry must be re-cached; another matching live relation in the same
+ * plan can still trigger re-caching.
+ */
+ def recacheByPath(
+ spark: SparkSession,
+ resourcePath: Path,
+ fs: FileSystem,
+ includeTimeTravel: Boolean): Unit = {
val qualifiedPath = fs.makeQualified(resourcePath)
- recacheByCondition(spark, _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath)))
+ recacheByCondition(
+ spark,
+ _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath, includeTimeTravel)))
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58907:
URL: https://github.com/apache/spark/pull/58907#issuecomment-5818223754
Thanks - I reviewed the updated mixed pinned/live cache regression. It now exercises the compound plan, rebuilds the cache through the live relation, and verifies that the pinned FileIndex is not refreshed.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5771754796","thread_id":"conversation:5771754796","verdict_sha256":"9359e3b7a2b997b6ed10920723b67e480d7702e03d520068d359980ef5539a29"} -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58907:
URL: https://github.com/apache/spark/pull/58907#discussion_r4068205385
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -582,8 +593,24 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* `HadoopFsRelation` node(s) as part of its logical plan.
*/
def recacheByPath(spark: SparkSession, resourcePath: Path, fs: FileSystem): Unit = {
+ recacheByPath(spark, resourcePath, fs, includeTimeTravel = true)
+ }
+
+ /**
+ * Tries to re-cache matching entries whose logical plans contain a file-backed relation under
+ * `resourcePath`. If `includeTimeTravel` is false, immutable time-travel relations are ignored
+ * when deciding whether an entry must be re-cached; another matching live relation in the same
+ * plan can still trigger re-caching.
+ */
+ def recacheByPath(
+ spark: SparkSession,
+ resourcePath: Path,
+ fs: FileSystem,
+ includeTimeTravel: Boolean): Unit = {
val qualifiedPath = fs.makeQualified(resourcePath)
- recacheByCondition(spark, _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath)))
+ recacheByCondition(
+ spark,
+ _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath, includeTimeTravel)))
Review Comment:
**Non-blocking (P2):** Please add a cached join or union containing both matching pinned and live FileIndex relations. The regression should verify that write-driven invalidation refreshes the live index and rebuilds the compound cache while leaving the pinned index itself unrefreshed; the current single-relation cases would not catch an implementation that excludes the whole cache entry after seeing the immutable child.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58907:
URL: https://github.com/apache/spark/pull/58907#issuecomment-5818271624
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/269bae36567e9d8f8727b1e6b6c585c5f69909ad
- merged into branch-4.x https://github.com/apache/spark/commit/33fe09d94fe039a8ec9099d0b78ebe6933ac1128
- merged into branch-4.3 https://github.com/apache/spark/commit/6049ec7d1cee34e4e5f22c7c2760003ffc92dde9
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59631][SQL] Preserve cached V1 time-travel snapshots after table writes [spark]
Posted by "joelrobin18 (via GitHub)" <gi...@apache.org>.
joelrobin18 commented on code in PR #58907:
URL: https://github.com/apache/spark/pull/58907#discussion_r4059735990
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -582,8 +593,22 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* `HadoopFsRelation` node(s) as part of its logical plan.
*/
def recacheByPath(spark: SparkSession, resourcePath: Path, fs: FileSystem): Unit = {
+ recacheByPath(spark, resourcePath, fs, includeTimeTravel = true)
+ }
+
+ /**
+ * Tries to re-cache all the cache entries that contain `resourcePath` in one or more
Review Comment:
Updated the overload Scaladoc to define the flag precisely: when false, immutable time-travel relations are ignored when deciding whether to recache, while another matching live relation in the same cached plan can still trigger recaching. Addressed in ae900d93bde.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala:
##########
@@ -209,7 +209,11 @@ case class InsertIntoHadoopFsRelationCommand(
// refresh cached files in FileIndex
fileIndex.foreach(_.refresh())
// refresh data cache if table is cached
- sparkSession.sharedState.cacheManager.recacheByPath(sparkSession, outputPath, fs)
+ sparkSession.sharedState.cacheManager.recacheByPath(
+ sparkSession,
+ outputPath,
+ fs,
+ includeTimeTravel = false)
Review Comment:
Added a regression that executes a real Parquet append through the V1 write path. It materializes pinned and live caches, verifies that the production write preserves the pinned cache and clears the live cache, then confirms a fresh read sees the appended data. Addressed in ae900d93bde.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -592,16 +617,25 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
* in the plan. If found, we refresh the metadata and return true. Otherwise, this method returns
* false.
*/
- private def lookupAndRefresh(plan: LogicalPlan, fs: FileSystem, qualifiedPath: Path): Boolean = {
+ private def lookupAndRefresh(
+ plan: LogicalPlan,
+ fs: FileSystem,
+ qualifiedPath: Path,
+ includeTimeTravel: Boolean): Boolean = {
plan match {
case lr: LogicalRelation => lr.relation match {
case hr: HadoopFsRelation =>
- refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath)
+ refreshFileIndexIfNecessary(hr.location, fs, qualifiedPath, includeTimeTravel)
case _ => false
}
- case ExtractV2Table(fileTable: FileTable) =>
- refreshFileIndexIfNecessary(fileTable.fileIndex, fs, qualifiedPath)
+ case relation @ ExtractV2Table(fileTable: FileTable)
+ if includeTimeTravel || relation.timeTravelSpec.isEmpty =>
Review Comment:
Added path-based coverage using a real ParquetTable in a DataSourceV2Relation with a non-empty timeTravelSpec. Its FileIndex is explicitly live, so write-driven exclusion depends on the V2 guard; the test also verifies that explicit inclusive refresh clears the cache. Addressed in ae900d93bde.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org