spark-test commented on code in PR #59890:
URL: https://github.com/apache/spark/pull/59890#discussion_r4462516730
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -634,26 +636,45 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
fs: FileSystem,
includeTimeTravel: Boolean): Unit = {
val qualifiedPath = fs.makeQualified(resourcePath)
+ // recacheByCondition eagerly filters all entries before rebuilding any of
them, so an
+ // index shared across entries needs refreshing only once during this
call. Use object
+ // identity: equal root paths can have independent file-listing caches.
+ val refreshedIndexes = Collections.newSetFromMap(
+ new IdentityHashMap[FileIndex, java.lang.Boolean]())
recacheByCondition(
spark,
- _.plan.exists(lookupAndRefresh(_, fs, qualifiedPath, includeTimeTravel)))
+ cached => {
+ var refreshed = false
+ // lookupAndRefresh has side effects: exists would stop after the
first matching read.
+ // Refresh matching indexes in the operator tree before rebuilding a
multi-read cache.
+ // As before, this traversal does not enter plans held by subquery
expressions.
+ cached.plan.foreach { plan =>
Review Comment:
**Non-blocking (P2):** `TreeNode.foreach` follows only `children`, so this
walk never enters plans held by subquery expressions. In the normalized cache
key, `IN`, `EXISTS`, scalar, and lateral subquery plans stay inside
expressions, so their V1 `HadoopFsRelation` and V2 `FileTable` reads never
reach `lookupAndRefresh`: their indexes are not refreshed, and they cannot
select the entry for re-caching. The rebuild does not make up for it.
`tryRefreshPlan` and `V2TableRefreshUtil.refresh` reload only catalog-backed V2
tables, so these reads keep their existing index objects and file listings.
For example, with the default V1 parquet path:
```scala
spark.sql("SELECT * FROM parquet.`/d/a` WHERE id IN (SELECT id FROM
parquet.`/d/b`)").cache()
```
Add files under `/d/b` outside Spark's writer, as
`appendParquetWithoutRefresh` does, and call
`spark.catalog.refreshByPath("/d")`. The `/d/a` index is refreshed and the
entry is rebuilt, but the subquery keeps its old `/d/b` listing, so the cached
query silently misses the new rows. If files under `/d/b` were deleted instead,
reading the rebuilt cache fails with `FAILED_READ_FILE.FILE_NOT_EXIST`.
`refreshByPath("/d/b")`, or an append to `/d/b` through
`InsertIntoHadoopFsRelationCommand`, does not re-cache the entry at all. The
same two reads placed in a union are now refreshed correctly.
These reads hit the same stale-listing failure that this PR fixes for union
and join branches, and `refreshByPath` and `REFRESH` are documented to refresh
all cached data for any Dataset that contains the path. The previous round
offered documenting the narrower scope as one option, but documenting it still
leaves these cached queries returning stale rows.
**Recommended change:** In recacheByPath, walk each cached plan with
QueryPlan.foreachWithSubqueries instead of TreeNode.foreach. Keep the same
lookupAndRefresh call, the shared per-call refreshedIndexes set, and the
OR-accumulated match flag, and update the traversal comment so it no longer
says subquery plans are skipped. Add refreshByPath cache tests for file reads
inside subquery expressions, and update the PR description's scope statement to
match.
**Why this works:** foreachWithSubqueries applies the same function to each
node, then to the plans held by that node's subquery expressions, recursing
into their nodes. Every V1 HadoopFsRelation and V2 FileTable read in a cached
plan therefore reaches lookupAndRefresh. Each distinct index is still refreshed
once per call through refreshedIndexes, and a match inside a subquery selects
its entry exactly as an operator-tree match does. recacheByCondition rebuilds
selected entries around the same index objects, so the rebuilt plans read
refreshed listings in every branch and every subquery.
**Scope:** Extend the existing path-refresh traversal in CacheManager to
subquery plans, and add refreshByPath cache tests for reads held by subquery
expressions.
**Compatibility:** Entries with no matching read anywhere in the plan are
not rebuilt, unrelated indexes are never refreshed, each distinct index is
listed at most once per call, and time-travel relations stay excluded from
write-driven refresh.
**Risks:** After refreshByPath, REFRESH, or a V1 write under a path, cached
entries whose only read under that path sits inside a subquery expression are
now rebuilt. That is the intended correction, but it adds rebuild and
re-materialization work for those entries, and only reverting the traversal
change rolls it back. Every recacheByPath call, including the one after each V1
write, also walks the subquery plans of every cached entry. The added driver
work is one pattern match per subquery node, and only indexes under the path
are listed.
**Constraints:** Keep one per-call identity set shared by the operator tree
and subquery plans, so an index referenced in both is refreshed once per call.
Apply the time-travel exclusions to subquery-held reads exactly as to
operator-tree reads: includeTimeTravel, FileIndex.isTimeTravel, and the V2
relation's timeTravelSpec.
**Exclusions:**
- The children-only matching used by uncaching a table or view and by
re-caching by plan or by table name stays as it is: those operations refresh no
file listing and are separate from path-based refresh.
**Success:** After refreshByPath on a parent path, a cached query whose IN,
EXISTS, scalar, or lateral subquery reads files under that path through its own
index returns rows from the newly added files, including when an outer read is
also under the path. A cached entry whose only read under the refreshed or
written path sits inside a subquery expression is rebuilt and returns fresh
rows. Each distinct index is still refreshed at most once per call, and both
entry selection for operator-tree reads and the time-travel exclusions behave
as before.
**Verification:**
- **Regression:** Cache a query whose outer read and IN-subquery read are
separate reads under one parent directory, add files under both directories
outside Spark's writer, refresh the parent, and assert that the cached query
returns the complete fresh result; the same case must fail when the traversal
visits only operator children.
- **Regression:** Cache a query whose only read under the refreshed path is
inside a subquery expression, add files there outside Spark's writer, refresh
that path, and assert that the cached query returns fresh rows.
- **Behavior:** Run the subquery cases with both V1 and V2 file sources and
assert the analyzed relation kinds, and use a refresh-counting index to assert
that an index referenced both in the operator tree and in a subquery is
refreshed exactly once per call.
- **Compatibility:** Existing path-refresh coverage keeps passing:
multi-read refresh in the operator tree, partial matches, deduplication of
shared and of distinct equal indexes within one call, and exclusion of
time-travel relations from write-driven refresh.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -672,19 +694,21 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
/**
* Refresh the given [[FileIndex]] if one of its root paths is a
subdirectory of
* `qualifiedPath` and either `includeTimeTravel` is true or the index is
not an immutable
- * time-travel snapshot.
- * @return whether the [[FileIndex]] was refreshed.
+ * time-travel snapshot. Refresh each index at most once per recacheByPath
call.
+ * @return whether the [[FileIndex]] matches, including one already
refreshed during this call.
*/
private def refreshFileIndexIfNecessary(
fileIndex: FileIndex,
fs: FileSystem,
qualifiedPath: Path,
- includeTimeTravel: Boolean): Boolean = {
+ includeTimeTravel: Boolean,
+ refreshedIndexes: JSet[FileIndex]): Boolean = {
val needToRefresh = (includeTimeTravel || !fileIndex.isTimeTravel) &&
fileIndex.rootPaths
.map(_.makeQualified(fs.getUri, fs.getWorkingDirectory))
.exists(isSubDir(qualifiedPath, _))
- if (needToRefresh) fileIndex.refresh()
+ if (needToRefresh && refreshedIndexes.add(fileIndex)) fileIndex.refresh()
Review Comment:
**Nit (P3):** `needToRefresh` is computed before the set is consulted, so
every later occurrence of an index that this call has already refreshed
re-qualifies all of its root paths and walks `isSubDir` before `add()` returns
`false` and the refresh is skipped. An index enters `refreshedIndexes` only
after it matched, and `qualifiedPath`, `fs`, and `includeTimeTravel` are fixed
for the call, so membership already answers the question.
Within one plan this is new work. `df.union(df)`, or a self-join over one
read, now matches the same index once per occurrence, where `exists` stopped
after the first. Every other cached entry that shares the index repeats the
match as well. The cost is driver-side `Path` work. It is negligible for
single-path reads, but it grows with the root-path count of globbed or
multi-path reads. Results and refresh counts are unaffected.
Checking the set first gives the same results and matches the intent of the
comment below:
```scala
refreshedIndexes.contains(fileIndex) || {
val needToRefresh = (includeTimeTravel || !fileIndex.isTimeTravel) &&
fileIndex.rootPaths
.map(_.makeQualified(fs.getUri, fs.getWorkingDirectory))
.exists(isSubDir(qualifiedPath, _))
if (needToRefresh) {
refreshedIndexes.add(fileIndex)
fileIndex.refresh()
}
needToRefresh
}
```
The V2 `timeTravelSpec` guard in `lookupAndRefresh` still runs before this
check, and the built-in indexes keep the same `rootPaths` and `isTimeTravel`
across `refresh()`.
**Verification:**
- **Compatibility:** Existing path-refresh coverage keeps the same refresh
counts and results for an index repeated within one cached plan, distinct
indexes with equal root paths, an index shared by several cached entries, a
partially matching plan with an unrelated read, and time-travel relations
excluded from write-driven refresh.
- **Inspection:** The refreshed-index membership check runs before root-path
qualification and the isSubDir walk, so a repeated occurrence of an already
refreshed index performs no Path work.
--
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]
