spark-test commented on code in PR #59890:
URL: https://github.com/apache/spark/pull/59890#discussion_r4419772209
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -649,11 +666,13 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
plan: LogicalPlan,
fs: FileSystem,
qualifiedPath: Path,
- includeTimeTravel: Boolean): Boolean = {
+ includeTimeTravel: Boolean,
+ refreshedIndexes: JSet[FileIndex]): Boolean = {
Review Comment:
**Nit (P3):** The ScalaDoc just above this method (lines 659-664) predates
this change and no longer describes it. With the new `refreshedIndexes`
parameter, a matching index that was already refreshed earlier in the same
`recacheByPath` call makes this method return `true` without calling
`refresh()` (`refreshFileIndexIfNecessary`, lines 708-710). The
`first.union(second).union(first)` shared-index test hits exactly this case.
The doc still says "If found, we refresh the metadata and return true". It also
says the method traverses the plan, which the caller now does with `foreach`,
and it names only `HadoopFsRelation`, although the V2 `FileTable` arm is
handled too. A reader of this method's contract would expect every match to
re-list its index, and the deduplication rule is documented only on the private
callee.
`refreshFileIndexIfNecessary`'s doc was already updated for this change, so
something along the same lines would work here, for example:
```scala
/**
* Checks whether `plan` is a V1 `HadoopFsRelation` or V2 `FileTable` read
whose file index
* has a root path under `qualifiedPath`; time-travel snapshots are
skipped unless
* `includeTimeTravel` is true. A matching index is refreshed only the
first time it is seen
* during the current recacheByPath call, as tracked by `refreshedIndexes`.
* @return whether the read matches, including one whose index was already
refreshed.
*/
```
**Verification:**
- **Inspection:** The lookupAndRefresh documentation no longer claims that
every match refreshes its index. It describes the single-node check, both the
V1 and V2 file-relation arms, the per-call set of already refreshed indexes,
and the true result for an already refreshed match, in terms consistent with
the refreshFileIndexIfNecessary documentation.
##########
sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala:
##########
@@ -855,6 +856,133 @@ class CachedTableSuite extends SharedSparkSession
}
}
+ private def appendParquetWithoutRefresh(path: String, value: Long): Unit = {
+ withTempDir { staging =>
+ val output = new File(staging, "data")
+ spark.range(value, value + 1).write.parquet(output.getCanonicalPath)
+ // Move files externally so Spark does not refresh the destination
before refreshByPath.
+ output.listFiles().filter(_.getName.endsWith(".parquet")).foreach { file
=>
+ Files.move(file.toPath, new File(path, file.getName).toPath)
+ }
+ }
+ }
+
+ for {
+ useV1SourceList <- Seq("parquet", "")
+ relationCount <- Seq(1, 2, 3)
+ } {
+ val sourceVersion = if (useV1SourceList.nonEmpty) "V1" else "V2"
+ test(s"refreshByPath refreshes all $relationCount file indexes in one
cached plan " +
+ s"($sourceVersion)") {
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1SourceList) {
+ withTempDir { dir =>
+ val paths = (0 until relationCount).map(i => new File(dir,
s"part$i").getCanonicalPath)
+ paths.foreach(path => spark.range(1).write.parquet(path))
+ val cached =
paths.map(spark.read.parquet(_)).reduce(_.union(_)).cache()
+ try {
+ // Check the analyzed relations so a V2 fallback cannot silently
test only V1.
+ val fileSources = cached.queryExecution.analyzed.collect {
+ case LogicalRelationWithTable(_: HadoopFsRelation, _) => "V1"
+ case DataSourceV2Relation(_: FileTable, _, _, _, _, _) => "V2"
+ }
+ assert(fileSources == Seq.fill(relationCount)(sourceVersion))
+ assert(cached.count() == relationCount)
+ paths.foreach(path => appendParquetWithoutRefresh(path, 1L))
+ assert(cached.count() == relationCount)
+ spark.catalog.refreshByPath(dir.getCanonicalPath)
Review Comment:
**Nit (P3):** Every plan these tests cache ends its pre-order walk on a read
under the refreshed path. That covers the one- to three-read unions here,
`first.union(second).union(first)`, and the cross-entry unions, and it also
holds for the existing `refreshByPath`/`recacheByPath` tests, including the
pinned-then-live union in `V1TimeTravelCacheSuite`. So nothing checks that
entry selection in the new `foreach` loop (CacheManager.scala:647-655) is
independent of node order, which the replaced `exists` guaranteed by
construction. If the `if` were later rewritten as `refreshed =
lookupAndRefresh(...)`, the flag would reflect only the last node visited and
every test would still pass. A cached
`spark.read.parquet("/d/a").union(spark.read.parquet("/e/b"))` refreshed with
`refreshByPath("/d")` would then not be re-cached and would keep serving stale
`/d/a` rows. One case whose first read is under the refreshed directory and
whose last read is outside it would pin this.
**Verification:**
- **Regression:** Cache a union whose first read is under the refreshed
directory and whose last read is outside it, add files under both outside
Spark's writer, and call refreshByPath on the first directory. The cached query
must then return the first read's new rows, and the test must fail if entry
selection reflects only the last visited node.
- **Behavior:** In the same partially matching plan, the matching index is
refreshed exactly once per refreshByPath call and the index of the trailing
read outside the refreshed directory is not refreshed.
--
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]
