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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
