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: [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]
